15832450392

Committed 23 Jun 2025 06:44PM UTC coverage: 61.138%. First build

Build # 15832450392

Build Type

push

github

Committed by

web-flow

Commit Message

BL-255 | Add `tofu` command to cli (#305)

* Rename TerraformContainer to TFContainer

* Create tofu command

* Add tofu command to cli and make 'tf' run tofu instead of terraform

* Missing whitespace

* Fix assertion

* Black

* Fix version check to consider tofu images

* Rename terraform module to tf

* Fix references

* Rename tf test files

* Support `TF_IMAGE*` config in build.env

* Black

* Fix assert_called_once invocation in test

* Attach version subcommand to tf commands

* Missed some terraform references

* Fix typo

* Fix kubectl discovery issue

* Upgrade release drafter version

Coverage Stats

210 of 476 branches covered (44.12%)

Branch coverage included in aggregate %.

48 of 57 new or added lines in 8 files covered. (84.21%)

2606 of 4130 relevant lines covered (63.1%)

0.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

38.05

/leverage/modules/tf.py

import re
from pathlib import Path
from typing import Sequence

import click
from click.exceptions import Exit

from leverage import logger
from leverage._internals import pass_container, pass_state
from leverage._utils import ExitError, parse_tf_file
from leverage.container import TFContainer
from leverage.container import get_docker_client
from leverage.modules.utils import env_var_option, mount_option, auth_mfa, auth_sso

REGION = (
    r"global|(?:[a-z]{2}-(?:gov-)?"
    r"(?:central|north|south|east|west|northeast|northwest|southeast|southwest|secret|topsecret)-[1-4])"
)


# ###########################################################################
# CREATE THE TOFU AND TERRAFORM GROUPS
# ###########################################################################
@click.group()
@mount_option
@env_var_option
@pass_state
def tofu(state, env_var, mount):
    """Run OpenTofu commands in a custom containerized environment that provides extra functionality when interacting
    with your cloud provider such as handling multi factor authentication for you.
    All tofu subcommands that receive extra args will pass the given strings as is to their corresponding OpenTofu
    counterparts in the container. For example as in `leverage tofu apply -auto-approve` or
    `leverage tofu init -reconfigure`
    """
    if env_var:
        env_var = dict(env_var)

    state.container = TFContainer(get_docker_client(), mounts=mount, env_vars=env_var)
    state.container.ensure_image()


@click.group()
@mount_option
@env_var_option
@pass_state
def terraform(state, env_var, mount):
    """Run Terraform commands in a custom containerized environment that provides extra functionality when interacting
    with your cloud provider such as handling multi factor authentication for you.
    All terraform subcommands that receive extra args will pass the given strings as is to their corresponding Terraform
    counterparts in the container. For example as in `leverage terraform apply -auto-approve` or
    `leverage terraform init -reconfigure`
    """
    if env_var:
        env_var = dict(env_var)

    state.container = TFContainer(get_docker_client(), terraform=True, mounts=mount, env_vars=env_var)
    state.container.ensure_image()


CONTEXT_SETTINGS = {"ignore_unknown_options": True}

# ###########################################################################
# CREATE THE TF GROUP'S COMMANDS
# ###########################################################################
#
# --layers is a ordered comma separated list of layer names
# The layer names are the relative paths of those layers relative to the current directory
# e.g. if CLI is called from /home/user/project/management and this is the tree:
# home
# ├── user
# │   └── project
# │       └── management
# │           ├── global
# │           |   └── security-base
# │           |   └── sso
# │           └── us-east-1
# │               └── terraform-backend
#
# Then all three layers can be initialized as follows:
# leverage tf init --layers us-east-1/terraform-backend,global/security-base,global/sso
#
# It is an ordered list because the layers will be visited in the same order they were
# supplied.
#
layers_option = click.option(
    "--layers",
    type=str,
    default="",
    help="Layers to apply the action to. (an ordered, comma-separated list of layer names)",
)


@click.command(context_settings=CONTEXT_SETTINGS)
@click.option("--skip-validation", is_flag=True, help="Skip layout validation.")
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def init(context, tf: TFContainer, skip_validation, layers, args):
    """
    Initialize this layer.
    """
    invoke_for_all_commands(layers, _init, args, skip_validation)


@click.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def plan(context, tf, layers, args):
    """Generate an execution plan for this layer."""
    invoke_for_all_commands(layers, _plan, args)


@click.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def apply(context, tf, layers, args):
    """Build or change the infrastructure in this layer."""
    invoke_for_all_commands(layers, _apply, args)


@click.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def output(context, tf, layers, args):
    """Show all output variables of this layer."""
    invoke_for_all_commands(layers, _output, args)


@click.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def destroy(context, tf, layers, args):
    """Destroy infrastructure in this layer."""
    invoke_for_all_commands(layers, _destroy, args)


@click.command()
@pass_container
def version(tf):
    """Print version."""
    tf.disable_authentication()
    tf.start("version")


@click.command()
@auth_mfa
@auth_sso
@pass_container
def shell(tf, mfa, sso):
    """Open a shell into the Terraform container in this layer."""
    tf.disable_authentication()
    if sso:
        tf.enable_sso()

    if mfa:
        tf.enable_mfa()

    tf.start_shell()


@click.command("format", context_settings=CONTEXT_SETTINGS)
@click.argument("args", nargs=-1)
@pass_container
def _format(tf, args):
    """Check if all files meet the canonical format and rewrite them accordingly."""
    args = args if "-recursive" in args else (*args, "-recursive")
    tf.disable_authentication()
    tf.start("fmt", *args)


@click.command()
@pass_container
def validate(tf):
    """Validate code of the current directory. Previous initialization might be needed."""
    tf.disable_authentication()
    tf.start("validate")


@click.command("validate-layout")
@pass_container
def validate_layout(tf):
    """Validate layer conforms to Leverage convention."""
    tf.set_backend_key()
    return _validate_layout()


@click.command("import")
@click.argument("address")
@click.argument("_id", metavar="ID")
@pass_container
def _import(tf, address, _id):
    """Import a resource."""
    exit_code = tf.start_in_layer("import", *tf.tf_default_args, address, _id)

    if exit_code:
        raise Exit(exit_code)


@click.command("refresh-credentials")
@pass_container
def refresh_credentials(tf):
    """Refresh the AWS credentials used on the current layer."""
    tf.paths.check_for_layer_location()
    if exit_code := tf.refresh_credentials():
        raise Exit(exit_code)


# ###########################################################################
# ATTACH SUBCOMMANDS TO TF COMMANDS
# ###########################################################################

for subcommand in (
    init,
    plan,
    apply,
    output,
    destroy,
    version,
    shell,
    _format,
    validate,
    validate_layout,
    _import,
    refresh_credentials,
):
    tofu.add_command(subcommand)
    terraform.add_command(subcommand)


# ###########################################################################
# HANDLER FOR MANAGING THE BASE COMMANDS (init, plan, apply, destroy, output)
# ###########################################################################
@pass_container
def invoke_for_all_commands(tf, layers, command, args, skip_validation=True):
    """
    Invoke helper for "all" commands.

    Args:
        layers: comma separated value of relative layer path
            e.g.: global/security_audit,us-east-1/tf-backend
        command: init, plan, apply
    """

    # convert layers from string to list
    layers = layers.split(",") if len(layers) > 0 else []

    # based on the location type manage the layers parameter
    location_type = tf.paths.get_location_type()
    if location_type == "layer" and len(layers) == 0:
        # running on a layer
        layers = [tf.paths.cwd]
    elif location_type == "layer":
        # running on a layer but --layers was set
        raise ExitError(1, "Can not set [bold]--layers[/bold] inside a layer.")
    elif location_type in ["account", "layers-group"] and len(layers) == 0:
        # running on an account but --layers was not set
        raise ExitError(1, "[bold]--layers[/bold] has to be set.")
    elif location_type not in ["account", "layer", "layers-group"]:
        # running outside a layer and account
        raise ExitError(1, "This command has to be run inside a layer or account directory.")
    else:
        # running on an account with --layers set
        layers = [tf.paths.cwd / x for x in layers]

    # get current location
    original_location = tf.paths.cwd
    original_working_dir = tf.container_config["working_dir"]

    # validate each layer before calling the execute command
    for layer in layers:
        logger.debug(f"Checking for layer {layer}...")
        # change to current dir and set it in the container
        tf.paths.cwd = layer

        # check layers existence
        if not layer.is_dir():
            logger.error(f"Directory [red]{layer}[/red] does not exist or is not a directory\n")
            raise Exit(1)

        # set the s3 key
        tf.set_backend_key(skip_validation)

        # validate layer
        validate_for_all_commands(layer, skip_validation=skip_validation)

        # change to original dir and set it in the container
        tf.paths.cwd = original_location

    # check layers existence
    for layer in layers:
        if len(layers) > 1:
            logger.info(f"Invoking command for layer {layer}...")

        # change to current dir and set it in the container
        tf.paths.cwd = layer

        # set the working dir
        working_dir = f"{tf.paths.guest_base_path}/{tf.paths.cwd.relative_to(tf.paths.root_dir).as_posix()}"
        tf.container_config["working_dir"] = working_dir

        # execute the actual command
        command(args=args)

        # change to original dir and set it in the container
        tf.paths.cwd = original_location

        # change to original working dir
        tf.container_config["working_dir"] = original_working_dir

    return layers


def validate_for_all_commands(layer, skip_validation=False):
    """
    Validate existence of layer and, if set, all the Leverage related stuff
    of each of them

    Args:
        layer: a full layer directory
    """

    logger.debug(f"Checking layer {layer}...")
    if not skip_validation and not _validate_layout():
        logger.error(
            "Layer configuration doesn't seem to be valid. Exiting.\n"
            "If you are sure your configuration is actually correct "
            "you may skip this validation using the --skip-validation flag."
        )
        raise Exit(1)


# ###########################################################################
# BASE COMMAND EXECUTORS
# ###########################################################################
@pass_container
def _init(tf, args):
    """Initialize this layer."""

    args = [
        arg
        for index, arg in enumerate(args)
        if not arg.startswith("-backend-config") or not arg[index - 1] == "-backend-config"
    ]
    args.append(f"-backend-config={tf.paths.backend_tfvars}")

    tf.paths.check_for_layer_location()

    exit_code = tf.start_in_layer("init", *args)
    if exit_code:
        raise Exit(exit_code)


@pass_container
def _plan(tf, args):
    """Generate an execution plan for this layer."""
    exit_code = tf.start_in_layer("plan", *tf.tf_default_args, *args)

    if exit_code:
        raise Exit(exit_code)


def has_a_plan_file(args: Sequence[str]) -> bool:
    """Determine whether the list of arguments has a plan file at the end.

    Terraform apply arguments have the form "-target ADDRESS" or "-target=ADDRESS"
    in one case "-var 'NAME=value'" or "-var='NAME=value'". There are also flags
    with the form "-flag".
    We just need to know if there is or not a plan file as a last argument to
    decide if we prepend our default terraform arguments or not.

    Cases to consider:
     Args                                | Plan file present
    -------------------------------------|-------------------
     ()                                  | False
     ("-flag")                           | False
     ("-var=value")                      | False
     ("plan_file")                       | True
     (..., "-var", "value")              | False
     (..., "-flag", "plan_file")         | True
     (..., "-var=value", "plan_file")    | True
     (..., "-var", "value", "plan_file") | True

    """

    # Valid 'terraform apply' flags:
    # https://developer.hashicorp.com/terraform/cli/commands/apply
    tf_flags = [
        "-destroy",
        "-refresh-only",
        "-detailed-exitcode",
        "-auto-approve",
        "-compact-warnings",
        "-json",
        "-no-color",
    ]

    if not args or args[-1].startswith("-"):
        return False

    if len(args) > 1:
        second_last = args[-2]
        if second_last.startswith("-"):
            if not "=" in second_last and second_last not in tf_flags:
                return False

    return True


@pass_container
def _apply(tf, args: Sequence[str]) -> None:
    """Build or change the infrastructure in this layer."""
    default_args = [] if has_a_plan_file(args) else tf.tf_default_args
    logger.debug(f"Default args passed to apply command: {default_args}")

    exit_code = tf.start_in_layer("apply", *default_args, *args)

    if exit_code:
        logger.error(f"Command execution failed with exit code: {exit_code}")
        raise Exit(exit_code)


@pass_container
def _output(tf, args):
    """Show all output variables of this layer."""
    tf.start_in_layer("output", *args)


@pass_container
def _destroy(tf, args):
    """Destroy infrastructure in this layer."""
    exit_code = tf.start_in_layer("destroy", *tf.tf_default_args, *args)

    if exit_code:
        raise Exit(exit_code)


# ###########################################################################
# MISC FUNCTIONS
# ###########################################################################
def _make_layer_backend_key(cwd, account_dir, account_name):
    """Create expected backend key.

    Args:
        cwd (pathlib.Path): Current Working Directory (Layer Directory)
        account_dir (pathlib.Path): Account Directory
        account_name (str): Account Name

    Returns:
        list of lists: Backend bucket key parts
    """
    resp = []

    layer_path = cwd.relative_to(account_dir)
    layer_path = layer_path.as_posix().split("/")
    # Check region directory to keep retro compat
    if re.match(REGION, layer_path[0]):
        layer_paths = [layer_path[1:], layer_path]
    else:
        layer_paths = [layer_path]

    curated_layer_paths = []
    # Remove layer name prefix
    for layer_path in layer_paths:
        curated_layer_path = []
        for lp in layer_path:
            if lp.startswith("base-"):
                lp = lp.replace("base-", "")
            elif lp.startswith("tools-"):
                lp = lp.replace("tools-", "")
            curated_layer_path.append(lp)
        curated_layer_paths.append(curated_layer_path)

    curated_layer_paths_retrocomp = []
    for layer_path in curated_layer_paths:
        curated_layer_paths_retrocomp.append(layer_path)
        # check for tf/terraform variants
        for idx, lp in enumerate(layer_path):
            if lp.startswith("tf-"):
                layer_path_tmp = layer_path.copy()
                layer_path_tmp[idx] = layer_path_tmp[idx].replace("tf-", "terraform-")
                curated_layer_paths_retrocomp.append(layer_path_tmp)
                break
            elif lp.startswith("terraform-"):
                layer_path_tmp = layer_path.copy()
                layer_path_tmp[idx] = layer_path_tmp[idx].replace("terraform-", "tf-")
                curated_layer_paths_retrocomp.append(layer_path_tmp)
                break

    curated_layer_paths_withDR = []
    for layer_path in curated_layer_paths_retrocomp:
        curated_layer_paths_withDR.append(layer_path)
        curated_layer_path = []
        append_str = "-dr"
        for lp in layer_path:
            if re.match(REGION, lp):
                curated_layer_path.append(lp)
            else:
                curated_layer_path.append(lp + append_str)
                append_str = ""
        curated_layer_paths_withDR.append(curated_layer_path)

    for layer_path in curated_layer_paths_withDR:
        resp.append([account_name, *layer_path])

    return resp


@pass_container
def _validate_layout(tf: TFContainer):
    tf.paths.check_for_layer_location()

    # Check for `environment = <account name>` in account.tfvars
    account_name = tf.paths.account_conf.get("environment")
    logger.info("Checking environment name definition in [bold]account.tfvars[/bold]...")
    if account_name is None:
        logger.error("[red]✘ FAILED[/red]\n")
        raise Exit(1)
    logger.info("[green]✔ OK[/green]\n")

    # Check if account directory name matches with environment name
    if tf.paths.account_dir.stem != account_name:
        logger.warning(
            "[yellow]‼[/yellow] Account directory name does not match environment name.\n"
            f"  Expected [bold]{account_name}[/bold], found [bold]{tf.paths.account_dir.stem}[/bold]\n"
        )

    backend_key = tf.backend_key.split("/")

    # Flag to report layout validity
    valid_layout = True

    # Check backend bucket key
    expected_backend_keys = _make_layer_backend_key(tf.paths.cwd, tf.paths.account_dir, account_name)
    logger.info("Checking backend key...")
    logger.info(f"Found: '{'/'.join(backend_key)}'")
    backend_key = backend_key[:-1]

    if backend_key in expected_backend_keys:
        logger.info("[green]✔ OK[/green]\n")
    else:
        exp_message = [f"{'/'.join(x)}/terraform.tfstate" for x in expected_backend_keys]
        logger.info(f"Expected one of: {';'.join(exp_message)}")
        logger.error("[red]✘ FAILED[/red]\n")
        valid_layout = False

    backend_tfvars = Path(tf.paths.local_backend_tfvars)
    backend_tfvars = parse_tf_file(backend_tfvars) if backend_tfvars.exists() else {}

    logger.info("Checking [bold]backend.tfvars[/bold]:\n")
    names_prefix = f"{tf.project}-{account_name}"
    names_prefix_bootstrap = f"{tf.project}-bootstrap"

    # Check profile, bucket and dynamo table names:
    for field in ("profile", "bucket", "dynamodb_table"):
        logger.info(f"Checking if {field.replace('_', ' ')} starts with {names_prefix}...")
        if backend_tfvars.get(field, "").startswith(names_prefix) or (
            field == "profile" and backend_tfvars.get(field, "").startswith(names_prefix_bootstrap)
        ):
            logger.info("[green]✔ OK[/green]\n")
        else:
            logger.error("[red]✘ FAILED[/red]\n")
            valid_layout = False

    return valid_layout

1	import re	1✔
2	from pathlib import Path	1✔
3	from typing import Sequence	1✔
4
5	import click	1✔
6	from click.exceptions import Exit	1✔
7
8	from leverage import logger	1✔
9	from leverage._internals import pass_container, pass_state	1✔
10	from leverage._utils import ExitError, parse_tf_file	1✔
11	from leverage.container import TFContainer	1✔
12	from leverage.container import get_docker_client	1✔
13	from leverage.modules.utils import env_var_option, mount_option, auth_mfa, auth_sso	1✔
14
15	REGION = (	1✔
16	r"global\|(?:[a-z]{2}-(?:gov-)?"
17	r"(?:central\|north\|south\|east\|west\|northeast\|northwest\|southeast\|southwest\|secret\|topsecret)-[1-4])"
18	)
19
20
21	# ###########################################################################
22	# CREATE THE TOFU AND TERRAFORM GROUPS
23	# ###########################################################################
24	@click.group()	1✔
25	@mount_option	1✔
26	@env_var_option	1✔
27	@pass_state	1✔
28	def tofu(state, env_var, mount):	1✔
29	"""Run OpenTofu commands in a custom containerized environment that provides extra functionality when interacting
30	with your cloud provider such as handling multi factor authentication for you.
31	All tofu subcommands that receive extra args will pass the given strings as is to their corresponding OpenTofu
32	counterparts in the container. For example as in `leverage tofu apply -auto-approve` or
33	`leverage tofu init -reconfigure`
34	"""
NEW 35	if env_var:	×
NEW 36	env_var = dict(env_var)	×
37
NEW 38	state.container = TFContainer(get_docker_client(), mounts=mount, env_vars=env_var)	×
NEW 39	state.container.ensure_image()	×
40
41
42	@click.group()	1✔
43	@mount_option	1✔
44	@env_var_option	1✔
45	@pass_state	1✔
46	def terraform(state, env_var, mount):	1✔
47	"""Run Terraform commands in a custom containerized environment that provides extra functionality when interacting
48	with your cloud provider such as handling multi factor authentication for you.
49	All terraform subcommands that receive extra args will pass the given strings as is to their corresponding Terraform
50	counterparts in the container. For example as in `leverage terraform apply -auto-approve` or
51	`leverage terraform init -reconfigure`
52	"""
53	if env_var:	×
54	env_var = dict(env_var)	×
55
NEW 56	state.container = TFContainer(get_docker_client(), terraform=True, mounts=mount, env_vars=env_var)	×
57	state.container.ensure_image()	×
58
59
60	CONTEXT_SETTINGS = {"ignore_unknown_options": True}	1✔
61
62	# ###########################################################################
63	# CREATE THE TF GROUP'S COMMANDS
64	# ###########################################################################
65	#
66	# --layers is a ordered comma separated list of layer names
67	# The layer names are the relative paths of those layers relative to the current directory
68	# e.g. if CLI is called from /home/user/project/management and this is the tree:
69	# home
70	# ├── user
71	# │ └── project
72	# │ └── management
73	# │ ├── global
74	# │ \| └── security-base
75	# │ \| └── sso
76	# │ └── us-east-1
77	# │ └── terraform-backend
78	#
79	# Then all three layers can be initialized as follows:
80	# leverage tf init --layers us-east-1/terraform-backend,global/security-base,global/sso
81	#
82	# It is an ordered list because the layers will be visited in the same order they were
83	# supplied.
84	#
85	layers_option = click.option(	1✔
86	"--layers",
87	type=str,
88	default="",
89	help="Layers to apply the action to. (an ordered, comma-separated list of layer names)",
90	)
91
92
93	@click.command(context_settings=CONTEXT_SETTINGS)	1✔
94	@click.option("--skip-validation", is_flag=True, help="Skip layout validation.")	1✔
95	@layers_option	1✔
96	@click.argument("args", nargs=-1)	1✔
97	@pass_container	1✔
98	@click.pass_context	1✔
99	def init(context, tf: TFContainer, skip_validation, layers, args):	1✔
100	"""
101	Initialize this layer.
102	"""
103	invoke_for_all_commands(layers, _init, args, skip_validation)	×
104
105
106	@click.command(context_settings=CONTEXT_SETTINGS)	1✔
107	@layers_option	1✔
108	@click.argument("args", nargs=-1)	1✔
109	@pass_container	1✔
110	@click.pass_context	1✔
111	def plan(context, tf, layers, args):	1✔
112	"""Generate an execution plan for this layer."""
113	invoke_for_all_commands(layers, _plan, args)	×
114
115
116	@click.command(context_settings=CONTEXT_SETTINGS)	1✔
117	@layers_option	1✔
118	@click.argument("args", nargs=-1)	1✔
119	@pass_container	1✔
120	@click.pass_context	1✔
121	def apply(context, tf, layers, args):	1✔
122	"""Build or change the infrastructure in this layer."""
123	invoke_for_all_commands(layers, _apply, args)	×
124
125
126	@click.command(context_settings=CONTEXT_SETTINGS)	1✔
127	@layers_option	1✔
128	@click.argument("args", nargs=-1)	1✔
129	@pass_container	1✔
130	@click.pass_context	1✔
131	def output(context, tf, layers, args):	1✔
132	"""Show all output variables of this layer."""
133	invoke_for_all_commands(layers, _output, args)	×
134
135
136	@click.command(context_settings=CONTEXT_SETTINGS)	1✔
137	@layers_option	1✔
138	@click.argument("args", nargs=-1)	1✔
139	@pass_container	1✔
140	@click.pass_context	1✔
141	def destroy(context, tf, layers, args):	1✔
142	"""Destroy infrastructure in this layer."""
143	invoke_for_all_commands(layers, _destroy, args)	×
144
145
146	@click.command()	1✔
147	@pass_container	1✔
148	def version(tf):	1✔
149	"""Print version."""
150	tf.disable_authentication()	×
151	tf.start("version")	×
152
153
154	@click.command()	1✔
155	@auth_mfa	1✔
156	@auth_sso	1✔
157	@pass_container	1✔
158	def shell(tf, mfa, sso):	1✔
159	"""Open a shell into the Terraform container in this layer."""
160	tf.disable_authentication()	×
161	if sso:	×
162	tf.enable_sso()	×
163
164	if mfa:	×
165	tf.enable_mfa()	×
166
167	tf.start_shell()	×
168
169
170	@click.command("format", context_settings=CONTEXT_SETTINGS)	1✔
171	@click.argument("args", nargs=-1)	1✔
172	@pass_container	1✔
173	def _format(tf, args):	1✔
174	"""Check if all files meet the canonical format and rewrite them accordingly."""
175	args = args if "-recursive" in args else (*args, "-recursive")	×
176	tf.disable_authentication()	×
177	tf.start("fmt", *args)	×
178
179
180	@click.command()	1✔
181	@pass_container	1✔
182	def validate(tf):	1✔
183	"""Validate code of the current directory. Previous initialization might be needed."""
184	tf.disable_authentication()	×
185	tf.start("validate")	×
186
187
188	@click.command("validate-layout")	1✔
189	@pass_container	1✔
190	def validate_layout(tf):	1✔
191	"""Validate layer conforms to Leverage convention."""
192	tf.set_backend_key()	×
193	return _validate_layout()	×
194
195
196	@click.command("import")	1✔
197	@click.argument("address")	1✔
198	@click.argument("_id", metavar="ID")	1✔
199	@pass_container	1✔
200	def _import(tf, address, _id):	1✔
201	"""Import a resource."""
202	exit_code = tf.start_in_layer("import", *tf.tf_default_args, address, _id)	×
203
204	if exit_code:	×
205	raise Exit(exit_code)	×
206
207
208	@click.command("refresh-credentials")	1✔
209	@pass_container	1✔
210	def refresh_credentials(tf):	1✔
211	"""Refresh the AWS credentials used on the current layer."""
212	tf.paths.check_for_layer_location()	×
213	if exit_code := tf.refresh_credentials():	×
214	raise Exit(exit_code)	×
215
216
217	# ###########################################################################
218	# ATTACH SUBCOMMANDS TO TF COMMANDS
219	# ###########################################################################
220
221	for subcommand in (	1✔
222	init,
223	plan,
224	apply,
225	output,
226	destroy,
227	version,
228	shell,
229	_format,
230	validate,
231	validate_layout,
232	_import,
233	refresh_credentials,
234	):
235	tofu.add_command(subcommand)	1✔
236	terraform.add_command(subcommand)	1✔
237
238
239	# ###########################################################################
240	# HANDLER FOR MANAGING THE BASE COMMANDS (init, plan, apply, destroy, output)
241	# ###########################################################################
242	@pass_container	1✔
243	def invoke_for_all_commands(tf, layers, command, args, skip_validation=True):	1✔
244	"""
245	Invoke helper for "all" commands.
246
247	Args:
248	layers: comma separated value of relative layer path
249	e.g.: global/security_audit,us-east-1/tf-backend
250	command: init, plan, apply
251	"""
252
253	# convert layers from string to list
254	layers = layers.split(",") if len(layers) > 0 else []	×
255
256	# based on the location type manage the layers parameter
257	location_type = tf.paths.get_location_type()	×
258	if location_type == "layer" and len(layers) == 0:	×
259	# running on a layer
260	layers = [tf.paths.cwd]	×
261	elif location_type == "layer":	×
262	# running on a layer but --layers was set
263	raise ExitError(1, "Can not set [bold]--layers[/bold] inside a layer.")	×
264	elif location_type in ["account", "layers-group"] and len(layers) == 0:	×
265	# running on an account but --layers was not set
266	raise ExitError(1, "[bold]--layers[/bold] has to be set.")	×
267	elif location_type not in ["account", "layer", "layers-group"]:	×
268	# running outside a layer and account
269	raise ExitError(1, "This command has to be run inside a layer or account directory.")	×
270	else:
271	# running on an account with --layers set
272	layers = [tf.paths.cwd / x for x in layers]	×
273
274	# get current location
275	original_location = tf.paths.cwd	×
276	original_working_dir = tf.container_config["working_dir"]	×
277
278	# validate each layer before calling the execute command
279	for layer in layers:	×
280	logger.debug(f"Checking for layer {layer}...")	×
281	# change to current dir and set it in the container
282	tf.paths.cwd = layer	×
283
284	# check layers existence
285	if not layer.is_dir():	×
286	logger.error(f"Directory [red]{layer}[/red] does not exist or is not a directory\n")	×
287	raise Exit(1)	×
288
289	# set the s3 key
290	tf.set_backend_key(skip_validation)	×
291
292	# validate layer
293	validate_for_all_commands(layer, skip_validation=skip_validation)	×
294
295	# change to original dir and set it in the container
296	tf.paths.cwd = original_location	×
297
298	# check layers existence
299	for layer in layers:	×
300	if len(layers) > 1:	×
301	logger.info(f"Invoking command for layer {layer}...")	×
302
303	# change to current dir and set it in the container
304	tf.paths.cwd = layer	×
305
306	# set the working dir
307	working_dir = f"{tf.paths.guest_base_path}/{tf.paths.cwd.relative_to(tf.paths.root_dir).as_posix()}"	×
308	tf.container_config["working_dir"] = working_dir	×
309
310	# execute the actual command
311	command(args=args)	×
312
313	# change to original dir and set it in the container
314	tf.paths.cwd = original_location	×
315
316	# change to original working dir
317	tf.container_config["working_dir"] = original_working_dir	×
318
319	return layers	×
320
321
322	def validate_for_all_commands(layer, skip_validation=False):	1✔
323	"""
324	Validate existence of layer and, if set, all the Leverage related stuff
325	of each of them
326
327	Args:
328	layer: a full layer directory
329	"""
330
331	logger.debug(f"Checking layer {layer}...")	×
332	if not skip_validation and not _validate_layout():	×
333	logger.error(	×
334	"Layer configuration doesn't seem to be valid. Exiting.\n"
335	"If you are sure your configuration is actually correct "
336	"you may skip this validation using the --skip-validation flag."
337	)
338	raise Exit(1)	×
339
340
341	# ###########################################################################
342	# BASE COMMAND EXECUTORS
343	# ###########################################################################
344	@pass_container	1✔
345	def _init(tf, args):	1✔
346	"""Initialize this layer."""
347
348	args = [	1✔
349	arg
350	for index, arg in enumerate(args)
351	if not arg.startswith("-backend-config") or not arg[index - 1] == "-backend-config"
352	]
353	args.append(f"-backend-config={tf.paths.backend_tfvars}")	1✔
354
355	tf.paths.check_for_layer_location()	1✔
356
357	exit_code = tf.start_in_layer("init", *args)	1✔
358	if exit_code:	1✔
359	raise Exit(exit_code)	×
360
361
362	@pass_container	1✔
363	def _plan(tf, args):	1✔
364	"""Generate an execution plan for this layer."""
365	exit_code = tf.start_in_layer("plan", tf.tf_default_args, args)	×
366
367	if exit_code:	×
368	raise Exit(exit_code)	×
369
370
371	def has_a_plan_file(args: Sequence[str]) -> bool:	1✔
372	"""Determine whether the list of arguments has a plan file at the end.
373
374	Terraform apply arguments have the form "-target ADDRESS" or "-target=ADDRESS"
375	in one case "-var 'NAME=value'" or "-var='NAME=value'". There are also flags
376	with the form "-flag".
377	We just need to know if there is or not a plan file as a last argument to
378	decide if we prepend our default terraform arguments or not.
379
380	Cases to consider:
381	Args \| Plan file present
382	-------------------------------------\|-------------------
383	() \| False
384	("-flag") \| False
385	("-var=value") \| False
386	("plan_file") \| True
387	(..., "-var", "value") \| False
388	(..., "-flag", "plan_file") \| True
389	(..., "-var=value", "plan_file") \| True
390	(..., "-var", "value", "plan_file") \| True
391
392	"""
393
394	# Valid 'terraform apply' flags:
395	# https://developer.hashicorp.com/terraform/cli/commands/apply
396	tf_flags = [	1✔
397	"-destroy",
398	"-refresh-only",
399	"-detailed-exitcode",
400	"-auto-approve",
401	"-compact-warnings",
402	"-json",
403	"-no-color",
404	]
405
406	if not args or args[-1].startswith("-"):	1✔
407	return False	1✔
408
409	if len(args) > 1:	1✔
410	second_last = args[-2]	1✔
411	if second_last.startswith("-"):	1✔
412	if not "=" in second_last and second_last not in tf_flags:	1✔
413	return False	1✔
414
415	return True	1✔
416
417
418	@pass_container	1✔
419	def _apply(tf, args: Sequence[str]) -> None:	1✔
420	"""Build or change the infrastructure in this layer."""
421	default_args = [] if has_a_plan_file(args) else tf.tf_default_args	×
422	logger.debug(f"Default args passed to apply command: {default_args}")	×
423
424	exit_code = tf.start_in_layer("apply", default_args, args)	×
425
426	if exit_code:	×
427	logger.error(f"Command execution failed with exit code: {exit_code}")	×
428	raise Exit(exit_code)	×
429
430
431	@pass_container	1✔
432	def _output(tf, args):	1✔
433	"""Show all output variables of this layer."""
434	tf.start_in_layer("output", *args)	×
435
436
437	@pass_container	1✔
438	def _destroy(tf, args):	1✔
439	"""Destroy infrastructure in this layer."""
440	exit_code = tf.start_in_layer("destroy", tf.tf_default_args, args)	×
441
442	if exit_code:	×
443	raise Exit(exit_code)	×
444
445
446	# ###########################################################################
447	# MISC FUNCTIONS
448	# ###########################################################################
449	def _make_layer_backend_key(cwd, account_dir, account_name):	1✔
450	"""Create expected backend key.
451
452	Args:
453	cwd (pathlib.Path): Current Working Directory (Layer Directory)
454	account_dir (pathlib.Path): Account Directory
455	account_name (str): Account Name
456
457	Returns:
458	list of lists: Backend bucket key parts
459	"""
460	resp = []	×
461
462	layer_path = cwd.relative_to(account_dir)	×
463	layer_path = layer_path.as_posix().split("/")	×
464	# Check region directory to keep retro compat
465	if re.match(REGION, layer_path[0]):	×
466	layer_paths = [layer_path[1:], layer_path]	×
467	else:
468	layer_paths = [layer_path]	×
469
470	curated_layer_paths = []	×
471	# Remove layer name prefix
472	for layer_path in layer_paths:	×
473	curated_layer_path = []	×
474	for lp in layer_path:	×
475	if lp.startswith("base-"):	×
476	lp = lp.replace("base-", "")	×
477	elif lp.startswith("tools-"):	×
478	lp = lp.replace("tools-", "")	×
479	curated_layer_path.append(lp)	×
480	curated_layer_paths.append(curated_layer_path)	×
481
482	curated_layer_paths_retrocomp = []	×
483	for layer_path in curated_layer_paths:	×
484	curated_layer_paths_retrocomp.append(layer_path)	×
485	# check for tf/terraform variants
486	for idx, lp in enumerate(layer_path):	×
487	if lp.startswith("tf-"):	×
488	layer_path_tmp = layer_path.copy()	×
489	layer_path_tmp[idx] = layer_path_tmp[idx].replace("tf-", "terraform-")	×
490	curated_layer_paths_retrocomp.append(layer_path_tmp)	×
491	break	×
492	elif lp.startswith("terraform-"):	×
493	layer_path_tmp = layer_path.copy()	×
494	layer_path_tmp[idx] = layer_path_tmp[idx].replace("terraform-", "tf-")	×
495	curated_layer_paths_retrocomp.append(layer_path_tmp)	×
496	break	×
497
498	curated_layer_paths_withDR = []	×
499	for layer_path in curated_layer_paths_retrocomp:	×
500	curated_layer_paths_withDR.append(layer_path)	×
501	curated_layer_path = []	×
502	append_str = "-dr"	×
503	for lp in layer_path:	×
504	if re.match(REGION, lp):	×
505	curated_layer_path.append(lp)	×
506	else:
507	curated_layer_path.append(lp + append_str)	×
508	append_str = ""	×
509	curated_layer_paths_withDR.append(curated_layer_path)	×
510
511	for layer_path in curated_layer_paths_withDR:	×
512	resp.append([account_name, *layer_path])	×
513
514	return resp	×
515
516
517	@pass_container	1✔
518	def _validate_layout(tf: TFContainer):	1✔
519	tf.paths.check_for_layer_location()	×
520
521	# Check for `environment = <account name>` in account.tfvars
522	account_name = tf.paths.account_conf.get("environment")	×
523	logger.info("Checking environment name definition in [bold]account.tfvars[/bold]...")	×
524	if account_name is None:	×
525	logger.error("[red]✘ FAILED[/red]\n")	×
526	raise Exit(1)	×
527	logger.info("[green]✔ OK[/green]\n")	×
528
529	# Check if account directory name matches with environment name
530	if tf.paths.account_dir.stem != account_name:	×
531	logger.warning(	×
532	"[yellow]‼[/yellow] Account directory name does not match environment name.\n"
533	f" Expected [bold]{account_name}[/bold], found [bold]{tf.paths.account_dir.stem}[/bold]\n"
534	)
535
536	backend_key = tf.backend_key.split("/")	×
537
538	# Flag to report layout validity
539	valid_layout = True	×
540
541	# Check backend bucket key
542	expected_backend_keys = _make_layer_backend_key(tf.paths.cwd, tf.paths.account_dir, account_name)	×
543	logger.info("Checking backend key...")	×
544	logger.info(f"Found: '{'/'.join(backend_key)}'")	×
545	backend_key = backend_key[:-1]	×
546
547	if backend_key in expected_backend_keys:	×
548	logger.info("[green]✔ OK[/green]\n")	×
549	else:
550	exp_message = [f"{'/'.join(x)}/terraform.tfstate" for x in expected_backend_keys]	×
551	logger.info(f"Expected one of: {';'.join(exp_message)}")	×
552	logger.error("[red]✘ FAILED[/red]\n")	×
553	valid_layout = False	×
554
555	backend_tfvars = Path(tf.paths.local_backend_tfvars)	×
556	backend_tfvars = parse_tf_file(backend_tfvars) if backend_tfvars.exists() else {}	×
557
558	logger.info("Checking [bold]backend.tfvars[/bold]:\n")	×
559	names_prefix = f"{tf.project}-{account_name}"	×
560	names_prefix_bootstrap = f"{tf.project}-bootstrap"	×
561
562	# Check profile, bucket and dynamo table names:
563	for field in ("profile", "bucket", "dynamodb_table"):	×
564	logger.info(f"Checking if {field.replace('_', ' ')} starts with {names_prefix}...")	×
565	if backend_tfvars.get(field, "").startswith(names_prefix) or (	×
566	field == "profile" and backend_tfvars.get(field, "").startswith(names_prefix_bootstrap)
567	):
568	logger.info("[green]✔ OK[/green]\n")	×
569	else:
570	logger.error("[red]✘ FAILED[/red]\n")	×
571	valid_layout = False	×
572
573	return valid_layout	×

binbashar / leverage / 15832450392

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous