10762575514

Committed 08 Sep 2024 07:34PM UTC coverage: 59.74% (-9.3%) from 69.031%

Build # 10762575514

Build Type

Pull #269

github

Committed by

Franr

Commit Message

fix test references

Pull Request Pull Request #269: [BV-195] Run toolbox with the host user

Run Details

204 of 508 branches covered (40.16%)

Branch coverage included in aggregate %.

63 of 67 new or added lines in 4 files covered. (94.03%)

4 existing lines in 1 file now uncovered.

2458 of 3948 relevant lines covered (62.26%)

0.62 hits per line

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

37.15

/leverage/modules/terraform.py

import re

import click
import hcl2
from click.exceptions import Exit

from leverage import logger
from leverage._internals import pass_container, pass_state
from leverage._utils import ExitError
from leverage.container import TerraformContainer
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 TERRAFORM GROUP
# ###########################################################################
@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 = TerraformContainer(get_docker_client(), mounts=mount, env_vars=env_var)
    state.container.ensure_image()


CONTEXT_SETTINGS = {"ignore_unknown_options": True}

# ###########################################################################
# CREATE THE TERRAFORM 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)",
)


@terraform.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: TerraformContainer, skip_validation, layers, args):
    """
    Initialize this layer.
    """
    invoke_for_all_commands(layers, _init, args, skip_validation)


@terraform.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)


@terraform.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)


@terraform.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)


@terraform.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)


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


@terraform.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()


@terraform.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)


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


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


@terraform.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)


@terraform.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)


# ###########################################################################
# 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 workgindir
        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 handle_apply_arguments_parsing(args):
    """Parse and process the arguments for the 'apply' command."""
    # Initialize new_args to handle both '-key=value' and '-key value'
    new_args = []
    skip_next = False  # Flag to skip the next argument if it's part of '-key value'

    for i, arg in enumerate(args):
        if skip_next:
            skip_next = False  # Reset flag and skip this iteration
            continue

        if arg.startswith("-") and not arg.startswith("-var"):
            if i + 1 < len(args) and not args[i + 1].startswith("-"):
                # Detected '-key value' pair; append them without merging
                new_args.append(arg)
                new_args.append(args[i + 1])
                skip_next = True  # Mark to skip the next item as it's already processed
                logger.debug(f"Detected '-key value' pair: {arg}, {args[i + 1]}")
            else:
                # Either '-key=value' or a standalone '-key'; just append
                new_args.append(arg)
                logger.debug(f"Appending standard -key=value or standalone argument: {arg}")
        else:
            # Handles '-var' and non '-' starting arguments
            new_args.append(arg)
            logger.debug(f"Appending argument (non '-' or '-var'): {arg}")

    return new_args


@pass_container
def _apply(tf, args):
    """Build or change the infrastructure in this layer."""
    # if there is a plan, remove all "-var" from the default args
    # Preserve the original `-var` removal logic and modify tf_default_args if necessary
    tf_default_args = tf.tf_default_args
    for arg in args:
        if not arg.startswith("-"):
            tf_default_args = [arg for index, arg in enumerate(tf_default_args) if not arg.startswith("-var")]
            break

    # Process arguments using the new parsing logic
    processed_args = handle_apply_arguments_parsing(args)

    logger.debug(f"Original tf_default_args: {tf_default_args}")
    logger.debug(f"Processed argument list for execution: {processed_args}")

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

binbashar / leverage / 10762575514

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