13227561880

Committed 09 Feb 2025 04:26PM UTC coverage: 60.209% (+0.04%) from 60.173%

Build # 13227561880

Build Type

Pull #296

github

Committed by

Franr

Commit Message

fix test

Pull Request Pull Request #296: BL-294 | Catch and handle HCL parsing errors

Run Details

184 of 448 branches covered (41.07%)

Branch coverage included in aggregate %.

14 of 18 new or added lines in 3 files covered. (77.78%)

183 existing lines in 10 files now uncovered.

2464 of 3950 relevant lines covered (62.38%)

0.62 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.
    """
    with open(file) as f:
        try:
            parsed = hcl2.load(f)
        except lark.exceptions.UnexpectedInput:
            raise ExitError(1, f"There is a parsing error with the {f.name} file. Please review it.")
        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	with open(file) as f:	1✔
125	try:	1✔
126	parsed = hcl2.load(f)	1✔
NEW 127	except lark.exceptions.UnexpectedInput:	×
NEW 128	raise ExitError(1, f"There is a parsing error with the {f.name} file. Please review it.")	×
129	else:
130	return parsed	1✔
131
132
133	class ContainerSession:	1✔
134	"""
135	Handle the start/stop cycle of a container.
136	Useful when you need to keep your container alive to share context between multiple commands.
137	"""
138
139	def __init__(self, docker_client: DockerClient, container_data):	1✔
140	self.docker_client = docker_client	1✔
141	self.container_data = container_data	1✔
142
143	def __enter__(self) -> Container:	1✔
144	self.docker_client.api.start(self.container_data)	1✔
145	return self.docker_client.containers.get(self.container_data["Id"])	1✔
146
147	def __exit__(self, exc_type, exc_value, exc_tb):	1✔
148	self.docker_client.api.stop(self.container_data)	1✔
149	self.docker_client.api.remove_container(self.container_data)	1✔
150
151
152	def key_finder(d: dict, target: str, avoid: str = None):	1✔
153	"""
154	Iterate over a dict of dicts and/or lists of dicts, looking for a key with value "target".
155	Collect and return all the values that matches "target" as key.
156	"""
157	values = []	1✔
158
159	for key, value in d.items():	1✔
160	if isinstance(value, dict):	1✔
161	# not the target but a dict? keep iterating recursively
162	values.extend(key_finder(value, target, avoid))	1✔
163	elif isinstance(value, list):	1✔
164	# not a dict but a list? it must be a list of dicts, keep iterating recursively
165	for dict_ in [d_ for d_ in value if isinstance(d_, dict)]:	1✔
166	values.extend(key_finder(dict_, target, avoid))	1✔
167	elif key == target:	1✔
168	if avoid and avoid in value:	1✔
169	# we found a key but the value contains <avoid> so skip it
170	continue	1✔
171	# found the target key, store the value
172	return [value] # return it as an 1-item array to avoid .extend() to split the string	1✔
173
174	return values	1✔
175
176
177	def get_or_create_section(updater: ConfigUpdater, section_name: str):	1✔
178	if not updater.has_section(section_name):	1✔
179	updater.add_section(section_name)	×
180	# add_section doesn't return the section object, so we need to retrieve it either case
181	return updater.get_section(section_name)	1✔

binbashar / leverage / 13227561880

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