14578385646

Committed 20 Apr 2025 02:07AM UTC coverage: 60.662%. Remained the same

Build # 14578385646

Build Type

Pull #303

github

Committed by

angelofenoglio

Commit Message

Format

Pull Request Pull Request #303: BUG-301 | Fix parsing error in `locals.tf` and 'config.tf` files

Run Details

198 of 464 branches covered (42.67%)

Branch coverage included in aggregate %.

4 of 6 new or added lines in 1 file covered. (66.67%)

1 existing line in 1 file now uncovered.

2550 of 4066 relevant lines covered (62.72%)

0.63 hits per line

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

83.7

/leverage/_utils.py

"""
    General use utilities.
"""

from pathlib import Path
from subprocess import run
from subprocess import PIPE

import hcl2
import lark
from click.exceptions import Exit
from configupdater import ConfigUpdater
from docker import DockerClient
from docker.models.containers import Container

from leverage import logger


def clean_exception_traceback(exception):
    """Delete special local variables from all frames of an exception's traceback
    as to avoid polluting the output when displaying it.

    Args:
        exception (Exception): The exception which traceback needs to be cleaned.

    Return:
        Exception: The exception with a clean traceback.
    """
    locals_to_delete = [
        "__builtins__",
        "__cached__",
        "__doc__",
        "__file__",
        "__loader__",
        "__name__",
        "__package__",
        "__spec__",
    ]

    traceback = exception.__traceback__

    while traceback is not None:
        frame = traceback.tb_frame
        for key in locals_to_delete:
            try:
                del frame.f_locals[key]
            except KeyError:
                pass

        traceback = traceback.tb_next

    return exception


def git(command):
    """Run the given git command.

    Args:
        command (str): Complete git command with or without the binary name.
    """
    command = command.split()
    command = ["git"] + command if command[0] != "git" else command

    run(command, stdout=PIPE, stderr=PIPE, check=True)


class CustomEntryPoint:
    """
    Set a custom entrypoint on the container while entering the context.
    Once outside, return it to its original value.
    """

    def __init__(self, container, entrypoint):
        self.container = container
        self.old_entrypoint = container.entrypoint
        self.new_entrypoint = entrypoint

    def __enter__(self):
        self.container.entrypoint = self.new_entrypoint

    def __exit__(self, *args, **kwargs):
        self.container.entrypoint = self.old_entrypoint


class AwsCredsEntryPoint(CustomEntryPoint):
    """
    Fetching AWS credentials by setting the SSO/MFA entrypoints.
    """

    def __init__(self, container, override_entrypoint=None):
        auth_method = container.auth_method()

        new_entrypoint = f"{auth_method}{container.entrypoint if override_entrypoint is None else override_entrypoint}"
        super(AwsCredsEntryPoint, self).__init__(container, entrypoint=new_entrypoint)

    def __exit__(self, *args, **kwargs):
        super(AwsCredsEntryPoint, self).__exit__(*args, **kwargs)
        if self.container.mfa_enabled:
            self.container.environment.update(
                {
                    "AWS_SHARED_CREDENTIALS_FILE": self.container.environment["AWS_SHARED_CREDENTIALS_FILE"].replace(
                        ".aws", "tmp"
                    ),
                    "AWS_CONFIG_FILE": self.container.environment["AWS_CONFIG_FILE"].replace(".aws", "tmp"),
                }
            )


class ExitError(Exit):
    """
    Raise an Exit exception but also print an error description.
    """

    def __init__(self, exit_code: int, error_description: str):
        logger.error(error_description)
        super(ExitError, self).__init__(exit_code)


def parse_tf_file(file: Path):
    """
    Open and parse an HCL file.
    In case of a parsing error, raise a user-friendly error.
    """
    try:
        content = file.read_text()
        parsed = hcl2.loads(content)
    except lark.exceptions.UnexpectedInput as error:
        raise ExitError(
            1,
            f"Possible invalid expression in file {file.name} near line {error.line}, column {error.column}\n"
            f"{error.get_context(content)}",
        )
    else:
        return parsed


class ContainerSession:
    """
    Handle the start/stop cycle of a container.
    Useful when you need to keep your container alive to share context between multiple commands.
    """

    def __init__(self, docker_client: DockerClient, container_data):
        self.docker_client = docker_client
        self.container_data = container_data

    def __enter__(self) -> Container:
        self.docker_client.api.start(self.container_data)
        return self.docker_client.containers.get(self.container_data["Id"])

    def __exit__(self, exc_type, exc_value, exc_tb):
        self.docker_client.api.stop(self.container_data)
        self.docker_client.api.remove_container(self.container_data)


def key_finder(d: dict, target: str, avoid: str = None):
    """
    Iterate over a dict of dicts and/or lists of dicts, looking for a key with value "target".
    Collect and return all the values that matches "target" as key.
    """
    values = []

    for key, value in d.items():
        if isinstance(value, dict):
            # not the target but a dict? keep iterating recursively
            values.extend(key_finder(value, target, avoid))
        elif isinstance(value, list):
            # not a dict but a list? it must be a list of dicts, keep iterating recursively
            for dict_ in [d_ for d_ in value if isinstance(d_, dict)]:
                values.extend(key_finder(dict_, target, avoid))
        elif key == target:
            if avoid and avoid in value:
                # we found a key but the value contains <avoid> so skip it
                continue
            # found the target key, store the value
            return [value]  # return it as an 1-item array to avoid .extend() to split the string

    return values


def get_or_create_section(updater: ConfigUpdater, section_name: str):
    if not updater.has_section(section_name):
        updater.add_section(section_name)
    # add_section doesn't return the section object, so we need to retrieve it either case
    return updater.get_section(section_name)

1	"""
2	General use utilities.
3	"""
4
5	from pathlib import Path	1✔
6	from subprocess import run	1✔
7	from subprocess import PIPE	1✔
8
9	import hcl2	1✔
10	import lark	1✔
11	from click.exceptions import Exit	1✔
12	from configupdater import ConfigUpdater	1✔
13	from docker import DockerClient	1✔
14	from docker.models.containers import Container	1✔
15
16	from leverage import logger	1✔
17
18
19	def clean_exception_traceback(exception):	1✔
20	"""Delete special local variables from all frames of an exception's traceback
21	as to avoid polluting the output when displaying it.
22
23	Args:
24	exception (Exception): The exception which traceback needs to be cleaned.
25
26	Return:
27	Exception: The exception with a clean traceback.
28	"""
29	locals_to_delete = [	1✔
30	"__builtins__",
31	"__cached__",
32	"__doc__",
33	"__file__",
34	"__loader__",
35	"__name__",
36	"__package__",
37	"__spec__",
38	]
39
40	traceback = exception.__traceback__	1✔
41
42	while traceback is not None:	1✔
43	frame = traceback.tb_frame	×
44	for key in locals_to_delete:	×
45	try:	×
46	del frame.f_locals[key]	×
47	except KeyError:	×
48	pass	×
49
50	traceback = traceback.tb_next	×
51
52	return exception	1✔
53
54
55	def git(command):	1✔
56	"""Run the given git command.
57
58	Args:
59	command (str): Complete git command with or without the binary name.
60	"""
61	command = command.split()	×
62	command = ["git"] + command if command[0] != "git" else command	×
63
64	run(command, stdout=PIPE, stderr=PIPE, check=True)	×
65
66
67	class CustomEntryPoint:	1✔
68	"""
69	Set a custom entrypoint on the container while entering the context.
70	Once outside, return it to its original value.
71	"""
72
73	def __init__(self, container, entrypoint):	1✔
74	self.container = container	1✔
75	self.old_entrypoint = container.entrypoint	1✔
76	self.new_entrypoint = entrypoint	1✔
77
78	def __enter__(self):	1✔
79	self.container.entrypoint = self.new_entrypoint	1✔
80
81	def __exit__(self, args, *kwargs):	1✔
82	self.container.entrypoint = self.old_entrypoint	1✔
83
84
85	class AwsCredsEntryPoint(CustomEntryPoint):	1✔
86	"""
87	Fetching AWS credentials by setting the SSO/MFA entrypoints.
88	"""
89
90	def __init__(self, container, override_entrypoint=None):	1✔
91	auth_method = container.auth_method()	1✔
92
93	new_entrypoint = f"{auth_method}{container.entrypoint if override_entrypoint is None else override_entrypoint}"	1✔
94	super(AwsCredsEntryPoint, self).__init__(container, entrypoint=new_entrypoint)	1✔
95
96	def __exit__(self, args, *kwargs):	1✔
97	super(AwsCredsEntryPoint, self).__exit__(args, *kwargs)	1✔
98	if self.container.mfa_enabled:	1✔
99	self.container.environment.update(	×
100	{
101	"AWS_SHARED_CREDENTIALS_FILE": self.container.environment["AWS_SHARED_CREDENTIALS_FILE"].replace(
102	".aws", "tmp"
103	),
104	"AWS_CONFIG_FILE": self.container.environment["AWS_CONFIG_FILE"].replace(".aws", "tmp"),
105	}
106	)
107
108
109	class ExitError(Exit):	1✔
110	"""
111	Raise an Exit exception but also print an error description.
112	"""
113
114	def __init__(self, exit_code: int, error_description: str):	1✔
115	logger.error(error_description)	1✔
116	super(ExitError, self).__init__(exit_code)	1✔
117
118
119	def parse_tf_file(file: Path):	1✔
120	"""
121	Open and parse an HCL file.
122	In case of a parsing error, raise a user-friendly error.
123	"""
124	try:	1✔
125	content = file.read_text()	1✔
126	parsed = hcl2.loads(content)	1✔
NEW 127	except lark.exceptions.UnexpectedInput as error:	×
NEW 128	raise ExitError(	×
129	1,
130	f"Possible invalid expression in file {file.name} near line {error.line}, column {error.column}\n"
131	f"{error.get_context(content)}",
132	)
133	else:
134	return parsed	1✔
135
136
137	class ContainerSession:	1✔
138	"""
139	Handle the start/stop cycle of a container.
140	Useful when you need to keep your container alive to share context between multiple commands.
141	"""
142
143	def __init__(self, docker_client: DockerClient, container_data):	1✔
144	self.docker_client = docker_client	1✔
145	self.container_data = container_data	1✔
146
147	def __enter__(self) -> Container:	1✔
148	self.docker_client.api.start(self.container_data)	1✔
149	return self.docker_client.containers.get(self.container_data["Id"])	1✔
150
151	def __exit__(self, exc_type, exc_value, exc_tb):	1✔
152	self.docker_client.api.stop(self.container_data)	1✔
153	self.docker_client.api.remove_container(self.container_data)	1✔
154
155
156	def key_finder(d: dict, target: str, avoid: str = None):	1✔
157	"""
158	Iterate over a dict of dicts and/or lists of dicts, looking for a key with value "target".
159	Collect and return all the values that matches "target" as key.
160	"""
161	values = []	1✔
162
163	for key, value in d.items():	1✔
164	if isinstance(value, dict):	1✔
165	# not the target but a dict? keep iterating recursively
166	values.extend(key_finder(value, target, avoid))	1✔
167	elif isinstance(value, list):	1✔
168	# not a dict but a list? it must be a list of dicts, keep iterating recursively
169	for dict_ in [d_ for d_ in value if isinstance(d_, dict)]:	1✔
170	values.extend(key_finder(dict_, target, avoid))	1✔
171	elif key == target:	1✔
172	if avoid and avoid in value:	1✔
173	# we found a key but the value contains <avoid> so skip it
174	continue	1✔
175	# found the target key, store the value
176	return [value] # return it as an 1-item array to avoid .extend() to split the string	1✔
177
178	return values	1✔
179
180
181	def get_or_create_section(updater: ConfigUpdater, section_name: str):	1✔
182	if not updater.has_section(section_name):	1✔
183	updater.add_section(section_name)	×
184	# add_section doesn't return the section object, so we need to retrieve it either case
185	return updater.get_section(section_name)	1✔

binbashar / leverage / 14578385646

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