def _generate_plan_file()

in samcli/hook_packages/terraform/hooks/prepare/hook.py [0:0]

• 51 lines of code
• 10 McCabe index (conditional complexity)

def _generate_plan_file(skip_prepare_infra: bool, terraform_application_dir: str) -> dict:
    """
    Call the relevant Terraform commands to generate, load and return the Terraform plan file
    which the AWS SAM CLI will then parse to extract the fields required to run local emulators.

    Parameters
    ----------
    skip_prepare_infra: bool
            Flag to skip skip prepare hook if we already have the metadata file. Default is False.
    terraform_application_dir: str
            The path where the hook can find the TF application.
    Returns
    -------
    dict
        The Terraform plan file in JSON format
    """
    log_msg = (
        (
            "The option to skip infrastructure preparation was provided, but AWS SAM CLI could not find "
            f"the metadata file. Preparing anyways.{os.linesep}Initializing Terraform application"
        )
        if skip_prepare_infra
        else "Initializing Terraform application"
    )
    LOG.info(log_msg)
    try:
        invoke_subprocess_with_loading_pattern(
            command_args={
                "args": ["terraform", "init", "-input=false"],
                "cwd": terraform_application_dir,
            },
            is_running_terraform_command=True,
        )

        # get json output of terraform plan
        LOG.info("Creating terraform plan and getting JSON output")
        with osutils.tempfile_platform_independent() as temp_file:
            invoke_subprocess_with_loading_pattern(
                # input false to avoid SAM CLI to stuck in case if the
                # Terraform project expects input, and customer does not provide it.
                command_args={
                    "args": ["terraform", "plan", "-out", temp_file.name, "-input=false"],
                    "cwd": terraform_application_dir,
                },
                is_running_terraform_command=True,
            )

            result = run(
                ["terraform", "show", "-json", temp_file.name],
                check=True,
                capture_output=True,
                cwd=terraform_application_dir,
            )
    except CalledProcessError as e:
        stderr_output = str(e.stderr)

        # stderr can take on bytes or just be a plain string depending on terminal
        if isinstance(e.stderr, bytes):
            stderr_output = e.stderr.decode("utf-8")

        # one of the subprocess.run calls resulted in non-zero exit code or some OS error
        LOG.debug(
            "Error running terraform command: \n" "cmd: %s \n" "stdout: %s \n" "stderr: %s \n",
            e.cmd,
            e.stdout,
            stderr_output,
        )

        raise PrepareHookException(
            f"There was an error while preparing the Terraform application.\n{stderr_output}"
        ) from e
    except LoadingPatternError as e:
        if TF_CLOUD_EXCEPTION_MESSAGE in e.message:
            raise TerraformCloudException(TF_CLOUD_HELP_MESSAGE)
        raise PrepareHookException(f"Error occurred when invoking a process:\n{e}") from e

    return dict(json.loads(result.stdout))