class Context:
    """Click context object


    # Manage config parsing, transforming deployment process.
    """

    def __init__(self,
                 config_filename: str,
                 artifact_store: str,
                 pretty_printer: StackPrettyPrinter,
                 stack_selector: StackSelector,
                 boto3_profile: Boto3Profile,
                 builder,
                 ):
        # simple
        self._config_filename: str = config_filename
        self._artifact_store: str = artifact_store
        # complex
        self._pretty_printer: StackPrettyPrinter = pretty_printer
        self._stack_selector = stack_selector
        self._boto3_profile = boto3_profile
        # lazy-loaded
        self._deployments = None
        self._command_runner = None

        # internals
        self.__builder = builder
        self.__lock = threading.RLock()

    @property
    def config_filename(self):
        """Config file name"""
        return self._config_filename

    @property
    def stack_selector(self):
        """Stack selector"""
        return self._stack_selector

    @property
    def boto3_profile(self):
        """Boto3 session profiles"""
        return self._boto3_profile

    @property
    def verbosity(self):
        """Verbosity level"""
        return self._pretty_printer.verbosity

    @property
    def ppt(self):
        """CLI pretty printer"""
        return self._pretty_printer

    @property
    def deployments(self):
        """Stack deployment spec."""
        # XXX: Lazy-loading so we don't get error if user is just looking at help
        with self.__lock:
            if self._deployments is None:
                self._deployments = self.__builder.parse_config(self.config_filename)
        return self._deployments

    @property
    def runner(self):
        """Command runner."""
        with self.__lock:
            if self._command_runner is None:
                self._command_runner = self.__builder.build_runner(
                    self.boto3_profile,
                    self._artifact_store,
                    self.deployments,
                    self.stack_selector,
                    self.ppt
                )
        return self._command_runner @click.command(cls=MultiCommand)
@click.version_option(version=__version__)
@click.option('--install-completion', is_flag=True, callback=install_callback,
              expose_value=False,
              help='Automatically install completion for the current shell. Make sure '
                   'to have psutil installed.')
@click.option('-f', '--file',
              type=click.Path(exists=False, dir_okay=True),
              default=None,
              help='Specify an alternate stack configuration file, default is '
                   'cfn-cli.yml.')
@click.option('-s', '--stack', autocompletion=stack_auto_complete,
              type=click.STRING, default='*',
              help='Select stacks to operate on, defined by STAGE_NAME.STACK_NAME, '
                   'nix glob is supported to select multiple stacks. Default value is '
                   '"*", which means all stacks in all stages.')
@click.option('-p', '--profile', autocompletion=profile_auto_complete,
              type=click.STRING, default=None,
              help='Override AWS profile specified in the config file.  Warning: '
                   'Don\'t use this option on stacks in different accounts.')
@click.option('-r', '--region',
              type=click.STRING, default=None,
              help='Override AWS region specified in the config.  Warning: Don\'t use '
                   'this option on stacks in different regions.')
@click.option('-a', '--artifact-store',
              type=click.STRING, default='',
              help='Override artifact store specified in the config.  Artifact store is'
                   'the s3 bucket used to store packaged template resource.  Warning: '
                   'Don\'t use this option on stacks in different accounts & regions.')
@click.option('-v', '--verbose', count=True,
              help='Be more verbose, can be specified multiple times.')
@click.pass_context
def cli(ctx, file, stack, profile, region, artifact_store, verbose):
    """AWS CloudFormation CLI - The missing CLI for CloudFormation.

    Quickstart: use `cfn-cli generate` to generate a new project.

    cfn-cli operates on a single YAML based config file and can manages stacks
    across regions & accounts.  By default, cfn-cli will try to locate config file
    cfn-cli.yml in current directory, override this using -f option:

    \b
        cfn-cli -f some-other-config-file.yaml <command>

    If the config contains multiple stacks, they can be can be selected using
    full qualified stack name:

    \b
        cfn-cli -s StageName.StackName <command>

    Unix style globs is also supported when selecting stacks to operate on:
    \b
        cfn-cli -s Backend.* <command>

    By default, command operates on all stacks in every stages, with order specified
    in the config file.

    Options can also be specified using environment variables:

    \b
        CFN_STACK=StageName.StackName cfn-cli <command>
    """

    # Setup global logging level
    if verbose >= 2: verbose = 2  # cap at 2

    logger = logging.getLogger()
    logger.setLevel(VERBOSITY_LOGLEVEL_MAPPING[verbose])

    # Build context from command options
    options = Options(
        config_filename=file,
        stack_selector=stack,
        profile_name=profile,
        region_name=region,
        artifact_store=artifact_store,
        verbosity=verbose,
    )
    context: Context = CONTEXT_BUILDER(options).build()

    # Assign context object
    ctx.obj = context class StackDeleteCommand(Command):
    SKIP_UPDATE_REFERENCES = True

    def run(self, stack_context):
        # stack contexts
        session = stack_context.session
        parameters = stack_context.parameters
        metadata = stack_context.metadata

        self.ppt.pprint_stack_name(stack_context.stack_key,
                                   parameters['StackName'],
                                   'Deleting stack ')

        # create boto3 cfn resource
        cfn = session.resource('cloudformation')
        self.ppt.pprint_session(session)
        self.ppt.pprint_parameters(parameters)

        # call boto3
        stack = cfn.Stack(parameters['StackName'])
        try:
            update_termination_protection(
                session,
                parameters.pop('EnableTerminationProtection', None),
                parameters['StackName'],
                self.ppt)
            self.ppt.pprint_stack(stack)
            stack.delete()
        except Exception as ex:
            if self.options.ignore_missing and \
                    is_stack_does_not_exist_exception(ex):
                self.ppt.secho(str(ex), fg='red')
                return
            else:
                raise

        # wait until delete complete
        if self.options.no_wait:
            self.ppt.secho('Stack is being deleted.')
        else:
            self.ppt.wait_until_delete_complete(session, stack)
            self.ppt.secho('Stack delete complete.', fg='green') class StackDeleteOptions(namedtuple('StackDeleteOptions',
                                    ['no_wait',
                                     'ignore_missing'])):
    pass class StackDeployCommand(Command):

    def run(self, stack_context):
        # stack contexts
        session = stack_context.session
        parameters = stack_context.parameters
        metadata = stack_context.metadata

        # print qualified name
        self.ppt.pprint_stack_name(stack_context.stack_key,
                                   parameters['StackName'],
                                   'Deploying stack ')

        # create boto3 cfn resource
        cfn = session.resource('cloudformation')
        self.ppt.pprint_session(session)

        # packaging if necessary
        stack_context.run_packaging()

        # overwrite using cli parameters
        if self.options.on_failure is not None:
            parameters['OnFailure'] = self.options.on_failure
        if self.options.disable_rollback:
            parameters['DisableRollback'] = self.options.disable_rollback
        if self.options.timeout_in_minutes:
            parameters['TimeoutInMinutes'] = self.options.timeout_in_minutes

        self.ppt.pprint_parameters(parameters)

        # calling boto3...
        try:
            client_request_token = 'awscfncli-deploy-{}'.format(uuid.uuid1())
            stack = cfn.create_stack(
                ClientRequestToken=client_request_token,
                **parameters)
        except Exception as ex:
            # skip existing stack error
            if self.options.ignore_existing and \
                    is_stack_already_exists_exception(ex):
                self.ppt.secho(str(ex), fg='red')
                return
            else:
                raise

        self.ppt.pprint_stack(stack)

        # wait until deployment is complete
        if self.options.no_wait:
            self.ppt.secho('Stack deployment started.')
        else:
            self.ppt.wait_until_deploy_complete(session, stack)
            self.ppt.secho('Stack deployment complete.', fg='green') class StackDeployOptions(
    namedtuple('StackDeployOptions',
               [
                   'no_wait',
                   'on_failure',
                   'disable_rollback',
                   'timeout_in_minutes',
                   'ignore_existing'
               ])):
    pass class StackStatusCommand(Command):
    SKIP_UPDATE_REFERENCES = True

    def run(self, stack_context):
        # stack contexts
        session = stack_context.session
        parameters = stack_context.parameters
        metadata = stack_context.metadata

        self.ppt.pprint_stack_name(stack_context.stack_key,
                                   parameters['StackName'])
        # shortcut since dry run is already handled in cli package
        if self.options.dry_run:
            return

        # metadata and parameters only get printed when verbosity>0
        self.ppt.pprint_metadata(metadata)
        self.ppt.pprint_parameters(parameters)

        cfn = session.resource('cloudformation')
        stack = cfn.Stack(parameters['StackName'])

        try:
            stack.stack_status
        except botocore.exceptions.ClientError as ex:
            if is_stack_does_not_exist_exception(ex):
                # make a "dummy" stack object so prettyprint is happy
                stack = dummy_stack(parameters['StackName'], 'STACK_NOT_FOUND')
                self.ppt.pprint_stack(stack, status=True)
                return
            else:
                raise

        self.ppt.pprint_stack(stack, status=True)

        if self.options.stack_resources:
            self.ppt.pprint_stack_resources(stack)

        if self.options.stack_exports:
            self.ppt.pprint_stack_parameters(stack)
            self.ppt.pprint_stack_exports(stack, session)
            client = session.client('cloudformation') class StackStatusOptions(namedtuple('StackStatusOptions',
                                    ['dry_run', 'stack_resources',
                                     'stack_exports'])):
    pass class StackSyncCommand(Command):

    def run(self, stack_context):
        # stack contexts
        session = stack_context.session
        parameters = stack_context.parameters
        metadata = stack_context.metadata

        # print stack qualified name
        self.ppt.pprint_stack_name(
            stack_context.stack_key,
            parameters['StackName'],
            'Syncing stack '
        )
        self.ppt.pprint_session(session)

        if self.options.use_previous_template:
            parameters.pop('TemplateBody', None)
            parameters.pop('TemplateURL', None)
            parameters['UsePreviousTemplate'] = True
        else:
            stack_context.run_packaging()

        # create cfn client
        client = session.client('cloudformation')

        # generate a unique changeset name
        changeset_name = '%s-%s' % \
                         (parameters['StackName'], str(uuid.uuid1()))

        # get changeset type: CREATE or UPDATE
        changeset_type, is_new_stack = self.check_changeset_type(client,
                                                                 parameters)

        # prepare stack parameters
        parameters['ChangeSetName'] = changeset_name
        parameters['ChangeSetType'] = changeset_type
        parameters.pop('StackPolicyBody', None)
        parameters.pop('StackPolicyURL', None)
        termination_protection = parameters.pop(
            'EnableTerminationProtection', None)

        self.ppt.pprint_parameters(parameters)

        # create changeset
        echo_pair('ChangeSet Name', changeset_name)
        echo_pair('ChangeSet Type', changeset_type)

        result = self.create_change_set(client, parameters)
        changeset_id = result['Id']
        echo_pair('ChangeSet ARN', changeset_id)

        self.ppt.wait_until_changset_complete(client, changeset_id)

        result = self.describe_change_set(client, changeset_name, parameters)
        self.ppt.pprint_changeset(result)

        # termination protection should be set after the creation of stack
        # or changeset
        update_termination_protection(session,
                                      termination_protection,
                                      parameters['StackName'],
                                      self.ppt)

        # check whether changeset is executable
        if result['Status'] not in ('AVAILABLE', 'CREATE_COMPLETE'):
            self.ppt.secho('ChangeSet not executable.', fg='red')
            return

        if self.options.confirm:
            if self.options.no_wait:
                return 
            if not self.ppt.confirm('Do you want to execute ChangeSet?'):
                return

        client_request_token = 'awscfncli-sync-{}'.format(uuid.uuid1())
        self.ppt.secho('Executing ChangeSet...')
        client.execute_change_set(
            ChangeSetName=changeset_name,
            StackName=parameters['StackName'],
            ClientRequestToken=client_request_token
        )

        cfn = session.resource('cloudformation')
        stack = cfn.Stack(parameters['StackName'])
        if self.options.no_wait:
            self.ppt.secho('ChangeSet execution started.')
        else:
            if is_new_stack:
                self.ppt.wait_until_deploy_complete(session, stack, self.options.disable_tail_events)
            else:
                self.ppt.wait_until_update_complete(session, stack, self.options.disable_tail_events)
            self.ppt.secho('ChangeSet execution complete.', fg='green')

    @backoff.on_exception(backoff.expo, botocore.exceptions.ClientError, max_tries=10,
                          giveup=is_not_rate_limited_exception)
    def create_change_set(self, client, parameters):
        return client.create_change_set(**parameters)

    @backoff.on_exception(backoff.expo, botocore.exceptions.ClientError, max_tries=10,
                          giveup=is_not_rate_limited_exception)
    def describe_change_set(self, client, changeset_name, parameters):
        return client.describe_change_set(
            ChangeSetName=changeset_name,
            StackName=parameters['StackName'],
        )

    @backoff.on_exception(backoff.expo, botocore.exceptions.ClientError, max_tries=10,
                          giveup=is_not_rate_limited_exception)
    def check_changeset_type(self, client, parameters):
        try:
            # check whether stack is already created.
            status = client.describe_stacks(StackName=parameters['StackName'])
            stack_status = status['Stacks'][0]['StackStatus']
        except botocore.exceptions.ClientError as e:

            if is_rate_limited_exception(e):
                # stack might exist but we got Throttling error, retry is needed so rerasing exception
                raise
            # stack not yet created
            is_new_stack = True
            changeset_type = 'CREATE'
        else:
            if stack_status == 'REVIEW_IN_PROGRESS':
                # first ChangeSet execution failed, create "new stack" changeset again
                is_new_stack = True
                changeset_type = 'CREATE'
            else:
                # updating an existing stack
                is_new_stack = False
                changeset_type = 'UPDATE'
        return changeset_type, is_new_stack class StackSyncOptions(namedtuple('StackSyncOptions',
                                  ['no_wait',
                                   'confirm',
                                   'use_previous_template',
                                   'disable_tail_events'])):
    pass class StackUpdateCommand(Command):

    def run(self, stack_context):
        # stack contexts
        session = stack_context.session
        parameters = stack_context.parameters
        metadata = stack_context.metadata

        # print stack qualified name
        self.ppt.pprint_stack_name(
            stack_context.stack_key,
            parameters['StackName'],
            'Updating stack '
        )

        # create boto3 cfn resource
        cfn = session.resource('cloudformation')
        self.ppt.pprint_session(session)

        # manipulate stack parameters for update call
        if self.options.use_previous_template:
            parameters.pop('TemplateBody', None)
            parameters.pop('TemplateURL', None)
            parameters['UsePreviousTemplate'] = True
        else:
            # packaging if necessary
            stack_context.run_packaging()

        parameters.pop('DisableRollback', None)
        parameters.pop('OnFailure', None)
        termination_protection = parameters.pop('EnableTerminationProtection',
                                                None)

        if self.options.override_policy is not None:
            self.ppt.secho(
                'Overriding stack policy with {} during update'.format(
                    self.options.override_policy), fg='red')
            parameters['StackPolicyDuringUpdateBody'] = \
                CANNED_STACK_POLICIES[self.options.override_policy]

        self.ppt.pprint_parameters(parameters)

        # termination protection state should be updated no matter
        # stack's update succeeded or not
        update_termination_protection(session,
                                      termination_protection,
                                      parameters['StackName'],
                                      self.ppt)

        # calling boto3...
        stack = cfn.Stack(parameters['StackName'])
        try:
            stack.update(**parameters)
        except Exception as ex:
            if self.options.ignore_no_update and \
                    is_no_updates_being_performed_exception(ex):
                self.ppt.secho(str(ex), fg='red')
                return
            else:
                raise

        self.ppt.pprint_stack(stack)

        # wait until update is complete
        if self.options.no_wait:
            self.ppt.secho('Stack update started.')
        else:
            self.ppt.wait_until_update_complete(session, stack)
            self.ppt.secho('Stack update complete.', fg='green') class StackUpdateOptions(namedtuple('StackUpdateOptions',
                                    ['no_wait',
                                     'use_previous_template',
                                     'ignore_no_update',
                                     'override_policy'])):
    pass
from unittest.mock import patch
from awscfncli2.cli.context import Context
from awscfncli2.cli.main import cli
from awscfncli2.runner.commands.stack_delete_command import (
    StackDeleteCommand,
    StackDeleteOptions,
)
from awscfncli2.runner.commands.stack_deploy_command import (
    StackDeployCommand,
    StackDeployOptions,
)
from awscfncli2.runner.commands.stack_status_command import (
    StackStatusCommand,
    StackStatusOptions,
)
from awscfncli2.runner.commands.stack_sync_command import (
    StackSyncCommand,
    StackSyncOptions,
)
from awscfncli2.runner.commands.stack_update_command import (
    StackUpdateCommand,
    StackUpdateOptions,
)
import click.testing
import pytest




@pytest.fixture()
def cli_runner():
    cli_runner = click.testing.CliRunner()
    with cli_runner.isolated_filesystem():
        cli_runner.invoke(cli, "generate")
        yield cli_runner


def test_stack_cancel_command(cli_runner):
    with patch.object(Context, "runner") as mock_command_runner:
        result = cli_runner.invoke(cli, "-f cfn-cli.yaml -s Develop.Table stack cancel")
        assert result.exit_code == 0


def test_stack_delete_command(cli_runner):
    with patch.object(Context, "runner") as mock_command_runner:
        result = cli_runner.invoke(
            cli,
            "-f cfn-cli.yaml -s Develop.Table stack delete "
            "--no-wait --ignore-missing",
        )
        assert result.exit_code == 0

        command_invoked = mock_command_runner.run.call_args[0][0]
        assert isinstance(command_invoked, StackDeleteCommand)

        options_collected = command_invoked.options
        assert isinstance(options_collected, StackDeleteOptions)
        assert options_collected._asdict() == {
            "no_wait": True,
            "ignore_missing": True,
        }


def test_stack_deploy_command(cli_runner):
    with patch.object(Context, "runner") as mock_command_runner:
        result = cli_runner.invoke(
            cli,
            "-f cfn-cli.yaml -s Develop.Table stack deploy "
            "--no-wait --on-failure=DO_NOTHING --disable-rollback "
            "--timeout-in-minutes=88 --ignore-existing",
        )
        assert result.exit_code == 0

        command_invoked = mock_command_runner.run.call_args[0][0]
        assert isinstance(command_invoked, StackDeployCommand)
        options_collected = command_invoked.options

        assert isinstance(options_collected, StackDeployOptions)
        assert options_collected._asdict() == {
            "no_wait": True,
            "on_failure": "DO_NOTHING",
            "disable_rollback": True,
            "timeout_in_minutes": 88,
            "ignore_existing": True,
        }


def test_status_command(cli_runner):
    with patch.object(Context, "runner") as mock_command_runner:
        result = cli_runner.invoke(cli, "-f cfn-cli.yaml -s Develop.Table status -re")

        assert result.exit_code == 0
        command_invoked = mock_command_runner.run.call_args[0][0]

        assert isinstance(command_invoked, StackStatusCommand)

        options_collected = command_invoked.options

        assert isinstance(options_collected, StackStatusOptions)