17239897634

Committed 26 Aug 2025 05:00AM UTC coverage: 61.145%. Remained the same

Build # 17239897634

Build Type

push

github

Committed by

web-flow

Commit Message

Fix | Remove reference to Terraform in error message (#312)

* Change error message for commands out of place

* Grammar

* Update lock file

* Fix integration test workflows

* Drop tofu image for now

* Small improvement in README

* Fix lock file

* Explicitly call terraform command

* Change error message for commands out of place

* Grammar

* Remove or adapt references to terraform

* Format

* Revert name of variable to render

* Pythonicism

Run Details

210 of 476 branches covered (44.12%)

Branch coverage included in aggregate %.

1 of 4 new or added lines in 3 files covered. (25.0%)

32 existing lines in 3 files now uncovered.

2610 of 4136 relevant lines covered (63.1%)

0.63 hits per line

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

71.75

/leverage/path.py

"""
    Utilities to obtain relevant files' and directories' locations
"""

import os
import pathlib
from pathlib import Path
from subprocess import CalledProcessError
from subprocess import PIPE
from subprocess import run

import hcl2

from leverage._utils import ExitError


class NotARepositoryError(RuntimeError):
    """When you are not running inside a git repository directory"""


def get_working_path():
    """Get the interpreters current directory.

    Returns:
        str: Current working directory.
    """
    return Path.cwd().as_posix()


def get_home_path():
    """Get the current user's home directory.

    Returns:
        str: User's home directory.
    """
    return Path.home().as_posix()


def get_root_path():
    """Get the path to the root of the Git repository.

    Raises:
        NotARepositoryError: If the current directory is not within a git repository.

    Returns:
        str: Root of the repository.
    """
    try:
        root = run(
            ["git", "rev-parse", "--show-toplevel"], stdout=PIPE, stderr=PIPE, check=True, encoding="utf-8"
        ).stdout

    except CalledProcessError as exc:
        if "fatal: not a git repository" in exc.stderr:
            raise NotARepositoryError("Not running in a git repository.")
    except FileNotFoundError as exc:
        raise NotARepositoryError("Not running in a git repository.")
    else:
        return root.strip()


def get_account_path():
    """Get the path to the current account directory.

    Returns:
        str: Path to the current account directory.
    """
    root_path = Path(get_root_path())
    cur_path = Path(get_working_path())
    prev_path = cur_path

    # NOTE: currently we only support up to 8 subdir levels. Normally we use
    #       only 2 subdirectories so this should be enough for most cases.
    for _ in range(8):
        if cur_path.resolve() == root_path:
            break

        prev_path = cur_path
        cur_path = cur_path.parent

    return prev_path.as_posix()


def get_global_config_path():
    """Get the path to the config that is common to all accounts.

    Returns:
        str: Global config file path.
    """
    return f"{get_root_path()}/config"


def get_account_config_path():
    """Get the path to the config of the current account.

    Returns:
        str: Current config file path.
    """
    return f"{get_account_path()}/config"


def get_build_script_path(filename="build.py"):
    """Get path to the build script containing all tasks to be run.
    Search through the current directory up to the repository's root directory.

    Args:
        filename (str, optional): The name of the build script containing the tasks.
            Defaults to "build.py".

    Returns:
        str: Build script file path. None if no file with the given name is found or
            the current directory is not a git repository.
    """
    try:
        root_path = Path(get_root_path())
    except NotARepositoryError:
        script = Path(filename)
        return script.absolute().as_posix() if script.exists() else None

    cur_path = Path(get_working_path())

    while True:
        build_script = list(cur_path.glob(filename))

        if build_script:
            return build_script[0].as_posix()

        if cur_path == root_path:
            break

        cur_path = cur_path.parent


class PathsHandler:
    COMMON_TF_VARS = "common.tfvars"
    ACCOUNT_TF_VARS = "account.tfvars"
    BACKEND_TF_VARS = "backend.tfvars"

    def __init__(self, env_conf: dict, container_user: str):
        self.container_user = container_user
        self.home = Path.home()
        self.cwd = Path.cwd()
        try:
            # TODO: just call get_root_path once and use it to initiate the rest of variables?
            self.root_dir = Path(get_root_path())
            self.account_dir = Path(get_account_path())
            self.common_config_dir = Path(get_global_config_path())
            self.account_config_dir = Path(get_account_config_path())
        except NotARepositoryError:
            raise ExitError(1, "Out of Leverage project context. Please cd into a Leverage project directory first.")

        # TODO: move the confs into a Config class
        common_config = self.common_config_dir / self.COMMON_TF_VARS
        self.common_conf = hcl2.loads(common_config.read_text()) if common_config.exists() else {}

        account_config = self.account_config_dir / self.ACCOUNT_TF_VARS
        self.account_conf = hcl2.loads(account_config.read_text()) if account_config.exists() else {}

        # Get project name
        self.project = self.common_conf.get("project", env_conf.get("PROJECT", False))
        if not self.project:
            raise ExitError(1, "Project name has not been set. Exiting.")

        # Project mount location
        self.project_long = self.common_conf.get("project_long", "project")
        self.guest_base_path = f"/{self.project_long}"

        # Ensure credentials directory
        self.host_aws_credentials_dir = self.home / ".aws" / self.project
        if not self.host_aws_credentials_dir.exists():
            self.host_aws_credentials_dir.mkdir(parents=True)
        self.sso_cache = self.host_aws_credentials_dir / "sso" / "cache"

    def update_cwd(self, new_cwd):
        self.cwd = new_cwd
        acc_folder = new_cwd.relative_to(self.root_dir).parts[0]

        self.account_config_dir = self.root_dir / acc_folder / "config"
        account_config_path = self.account_config_dir / self.ACCOUNT_TF_VARS
        self.account_conf = hcl2.loads(account_config_path.read_text())

    @property
    def guest_account_base_path(self):
        return f"{self.guest_base_path}/{self.account_dir.relative_to(self.root_dir).as_posix()}"

    @property
    def common_tfvars(self):
        return f"{self.guest_base_path}/config/{self.COMMON_TF_VARS}"

    @property
    def account_tfvars(self):
        return f"{self.guest_account_base_path}/config/{self.ACCOUNT_TF_VARS}"

    @property
    def backend_tfvars(self):
        return f"{self.guest_account_base_path}/config/{self.BACKEND_TF_VARS}"

    @property
    def guest_aws_credentials_dir(self):
        return str(f"/home/{self.container_user}/tmp" / Path(self.project))

    @property
    def host_aws_profiles_file(self):
        return f"{self.host_aws_credentials_dir}/config"

    @property
    def host_aws_credentials_file(self):
        return self.host_aws_credentials_dir / "credentials"

    @property
    def host_git_config_file(self):
        return self.home / ".gitconfig"

    @property
    def local_backend_tfvars(self):
        return self.account_config_dir / self.BACKEND_TF_VARS

    @property
    def sso_token_file(self):
        return f"{self.sso_cache}/token"

    def get_location_type(self):
        """
        Returns the location type:
        - root
        - account
        - config
        - layer
        - sublayer
        - not a project
        """
        if self.cwd == self.root_dir:
            return "root"
        elif self.cwd == self.account_dir:
            return "account"
        elif self.cwd in (self.common_config_dir, self.account_config_dir):
            return "config"
        elif (self.cwd.as_posix().find(self.account_dir.as_posix()) >= 0) and list(self.cwd.glob("*.tf")):
            return "layer"
        elif (self.cwd.as_posix().find(self.account_dir.as_posix()) >= 0) and not list(self.cwd.glob("*.tf")):
            return "layers-group"
        else:
            return "not a project"

    def assert_running_leverage_project(self):
        if self.root_dir == self.account_dir == self.common_config_dir == self.account_config_dir == self.cwd:
            raise ExitError(1, "Not running in a Leverage project. Exiting.")

    def guest_config_file(self, file):
        """Map config file in host to location in guest.

        Args:
            file (pathlib.Path): File in host to map

        Raises:
            Exit: If file is not contained in any valid config directory

        Returns:
            str: Path in guest to config file
        """
        file_name = file.name

        if file.parent == self.account_config_dir:
            return f"{self.guest_account_base_path}/config/{file_name}"
        if file.parent == self.common_config_dir:
            return f"{self.guest_base_path}/config/{file_name}"

        raise ExitError(1, "File is not part of any config directory.")

    @property
    def tf_cache_dir(self):
        return os.getenv("TF_PLUGIN_CACHE_DIR")

    def check_for_layer_location(self, path: Path = None):
        """Make sure the command is being run at layer level. If not, bail."""
        path = path or self.cwd
        if path in (self.common_config_dir, self.account_config_dir):
            raise ExitError(1, "Currently in a configuration directory, no OpenTofu/Terraform command can be run here.")

        if path in (self.root_dir, self.account_dir):
            raise ExitError(
                1,
                "This command cannot run neither in the root of the project or in" " the root directory of an account.",
            )

        if not list(path.glob("*.tf")):
            raise ExitError(1, "This command can only run at [bold]layer[/bold] level.")

    def check_for_cluster_layer(self, path: Path = None):
        path = path or self.cwd
        self.check_for_layer_location(path)
        # assuming the "cluster" layer will contain the expected EKS outputs
        if path.parts[-1] != "cluster":
            raise ExitError(1, "This command can only run at the [bold]cluster layer[/bold].")


def get_project_root_or_current_dir_path() -> Path:
    """Returns the project root if detected, otherwise the current path"""
    try:
        root = Path(get_root_path())
    except (NotARepositoryError, TypeError):
        root = Path.cwd()

    return root

1	"""
2	Utilities to obtain relevant files' and directories' locations
3	"""
4
5	import os	1✔
6	import pathlib	1✔
7	from pathlib import Path	1✔
8	from subprocess import CalledProcessError	1✔
9	from subprocess import PIPE	1✔
10	from subprocess import run	1✔
11
12	import hcl2	1✔
13
14	from leverage._utils import ExitError	1✔
15
16
17	class NotARepositoryError(RuntimeError):	1✔
18	"""When you are not running inside a git repository directory"""
19
20
21	def get_working_path():	1✔
22	"""Get the interpreters current directory.
23
24	Returns:
25	str: Current working directory.
26	"""
27	return Path.cwd().as_posix()	1✔
28
29
30	def get_home_path():	1✔
31	"""Get the current user's home directory.
32
33	Returns:
34	str: User's home directory.
35	"""
36	return Path.home().as_posix()	1✔
37
38
39	def get_root_path():	1✔
40	"""Get the path to the root of the Git repository.
41
42	Raises:
43	NotARepositoryError: If the current directory is not within a git repository.
44
45	Returns:
46	str: Root of the repository.
47	"""
48	try:	1✔
49	root = run(	1✔
50	["git", "rev-parse", "--show-toplevel"], stdout=PIPE, stderr=PIPE, check=True, encoding="utf-8"
51	).stdout
52
53	except CalledProcessError as exc:	1✔
54	if "fatal: not a git repository" in exc.stderr:	1✔
55	raise NotARepositoryError("Not running in a git repository.")	1✔
56	except FileNotFoundError as exc:	1✔
57	raise NotARepositoryError("Not running in a git repository.")	1✔
58	else:
59	return root.strip()	1✔
60
61
62	def get_account_path():	1✔
63	"""Get the path to the current account directory.
64
65	Returns:
66	str: Path to the current account directory.
67	"""
68	root_path = Path(get_root_path())	1✔
69	cur_path = Path(get_working_path())	1✔
70	prev_path = cur_path	1✔
71
72	# NOTE: currently we only support up to 8 subdir levels. Normally we use
73	# only 2 subdirectories so this should be enough for most cases.
74	for _ in range(8):	1✔
75	if cur_path.resolve() == root_path:	1✔
76	break	1✔
77
78	prev_path = cur_path	1✔
79	cur_path = cur_path.parent	1✔
80
81	return prev_path.as_posix()	1✔
82
83
84	def get_global_config_path():	1✔
85	"""Get the path to the config that is common to all accounts.
86
87	Returns:
88	str: Global config file path.
89	"""
90	return f"{get_root_path()}/config"	1✔
91
92
93	def get_account_config_path():	1✔
94	"""Get the path to the config of the current account.
95
96	Returns:
97	str: Current config file path.
98	"""
99	return f"{get_account_path()}/config"	1✔
100
101
102	def get_build_script_path(filename="build.py"):	1✔
103	"""Get path to the build script containing all tasks to be run.
104	Search through the current directory up to the repository's root directory.
105
106	Args:
107	filename (str, optional): The name of the build script containing the tasks.
108	Defaults to "build.py".
109
110	Returns:
111	str: Build script file path. None if no file with the given name is found or
112	the current directory is not a git repository.
113	"""
114	try:	1✔
115	root_path = Path(get_root_path())	1✔
116	except NotARepositoryError:	×
117	script = Path(filename)	×
118	return script.absolute().as_posix() if script.exists() else None	×
119
120	cur_path = Path(get_working_path())	1✔
121
122	while True:	1✔
123	build_script = list(cur_path.glob(filename))	1✔
124
125	if build_script:	1✔
126	return build_script[0].as_posix()	1✔
127
128	if cur_path == root_path:	1✔
129	break	1✔
130
131	cur_path = cur_path.parent	1✔
132
133
134	class PathsHandler:	1✔
135	COMMON_TF_VARS = "common.tfvars"	1✔
136	ACCOUNT_TF_VARS = "account.tfvars"	1✔
137	BACKEND_TF_VARS = "backend.tfvars"	1✔
138
139	def __init__(self, env_conf: dict, container_user: str):	1✔
140	self.container_user = container_user	1✔
141	self.home = Path.home()	1✔
142	self.cwd = Path.cwd()	1✔
143	try:	1✔
144	# TODO: just call get_root_path once and use it to initiate the rest of variables?
145	self.root_dir = Path(get_root_path())	1✔
146	self.account_dir = Path(get_account_path())	1✔
147	self.common_config_dir = Path(get_global_config_path())	1✔
148	self.account_config_dir = Path(get_account_config_path())	1✔
149	except NotARepositoryError:	×
150	raise ExitError(1, "Out of Leverage project context. Please cd into a Leverage project directory first.")	×
151
152	# TODO: move the confs into a Config class
153	common_config = self.common_config_dir / self.COMMON_TF_VARS	1✔
154	self.common_conf = hcl2.loads(common_config.read_text()) if common_config.exists() else {}	1✔
155
156	account_config = self.account_config_dir / self.ACCOUNT_TF_VARS	1✔
157	self.account_conf = hcl2.loads(account_config.read_text()) if account_config.exists() else {}	1✔
158
159	# Get project name
160	self.project = self.common_conf.get("project", env_conf.get("PROJECT", False))	1✔
161	if not self.project:	1✔
162	raise ExitError(1, "Project name has not been set. Exiting.")	×
163
164	# Project mount location
165	self.project_long = self.common_conf.get("project_long", "project")	1✔
166	self.guest_base_path = f"/{self.project_long}"	1✔
167
168	# Ensure credentials directory
169	self.host_aws_credentials_dir = self.home / ".aws" / self.project	1✔
170	if not self.host_aws_credentials_dir.exists():	1✔
171	self.host_aws_credentials_dir.mkdir(parents=True)	1✔
172	self.sso_cache = self.host_aws_credentials_dir / "sso" / "cache"	1✔
173
174	def update_cwd(self, new_cwd):	1✔
175	self.cwd = new_cwd	×
176	acc_folder = new_cwd.relative_to(self.root_dir).parts[0]	×
177
178	self.account_config_dir = self.root_dir / acc_folder / "config"	×
179	account_config_path = self.account_config_dir / self.ACCOUNT_TF_VARS	×
180	self.account_conf = hcl2.loads(account_config_path.read_text())	×
181
182	@property	1✔
183	def guest_account_base_path(self):	1✔
184	return f"{self.guest_base_path}/{self.account_dir.relative_to(self.root_dir).as_posix()}"	1✔
185
186	@property	1✔
187	def common_tfvars(self):	1✔
188	return f"{self.guest_base_path}/config/{self.COMMON_TF_VARS}"	1✔
189
190	@property	1✔
191	def account_tfvars(self):	1✔
192	return f"{self.guest_account_base_path}/config/{self.ACCOUNT_TF_VARS}"	1✔
193
194	@property	1✔
195	def backend_tfvars(self):	1✔
196	return f"{self.guest_account_base_path}/config/{self.BACKEND_TF_VARS}"	1✔
197
198	@property	1✔
199	def guest_aws_credentials_dir(self):	1✔
200	return str(f"/home/{self.container_user}/tmp" / Path(self.project))	1✔
201
202	@property	1✔
203	def host_aws_profiles_file(self):	1✔
204	return f"{self.host_aws_credentials_dir}/config"	×
205
206	@property	1✔
207	def host_aws_credentials_file(self):	1✔
208	return self.host_aws_credentials_dir / "credentials"	×
209
210	@property	1✔
211	def host_git_config_file(self):	1✔
212	return self.home / ".gitconfig"	1✔
213
214	@property	1✔
215	def local_backend_tfvars(self):	1✔
216	return self.account_config_dir / self.BACKEND_TF_VARS	×
217
218	@property	1✔
219	def sso_token_file(self):	1✔
220	return f"{self.sso_cache}/token"	×
221
222	def get_location_type(self):	1✔
223	"""
224	Returns the location type:
225	- root
226	- account
227	- config
228	- layer
229	- sublayer
230	- not a project
231	"""
232	if self.cwd == self.root_dir:	×
233	return "root"	×
234	elif self.cwd == self.account_dir:	×
235	return "account"	×
236	elif self.cwd in (self.common_config_dir, self.account_config_dir):	×
237	return "config"	×
238	elif (self.cwd.as_posix().find(self.account_dir.as_posix()) >= 0) and list(self.cwd.glob("*.tf")):	×
239	return "layer"	×
240	elif (self.cwd.as_posix().find(self.account_dir.as_posix()) >= 0) and not list(self.cwd.glob("*.tf")):	×
241	return "layers-group"	×
242	else:
243	return "not a project"	×
244
245	def assert_running_leverage_project(self):	1✔
246	if self.root_dir == self.account_dir == self.common_config_dir == self.account_config_dir == self.cwd:	1✔
247	raise ExitError(1, "Not running in a Leverage project. Exiting.")	×
248
249	def guest_config_file(self, file):	1✔
250	"""Map config file in host to location in guest.
251
252	Args:
253	file (pathlib.Path): File in host to map
254
255	Raises:
256	Exit: If file is not contained in any valid config directory
257
258	Returns:
259	str: Path in guest to config file
260	"""
261	file_name = file.name	×
262
263	if file.parent == self.account_config_dir:	×
264	return f"{self.guest_account_base_path}/config/{file_name}"	×
265	if file.parent == self.common_config_dir:	×
266	return f"{self.guest_base_path}/config/{file_name}"	×
267
268	raise ExitError(1, "File is not part of any config directory.")	×
269
270	@property	1✔
271	def tf_cache_dir(self):	1✔
272	return os.getenv("TF_PLUGIN_CACHE_DIR")	1✔
273
274	def check_for_layer_location(self, path: Path = None):	1✔
275	"""Make sure the command is being run at layer level. If not, bail."""
276	path = path or self.cwd	×
277	if path in (self.common_config_dir, self.account_config_dir):	×
NEW 278	raise ExitError(1, "Currently in a configuration directory, no OpenTofu/Terraform command can be run here.")	×
279
280	if path in (self.root_dir, self.account_dir):	×
281	raise ExitError(	×
282	1,
283	"This command cannot run neither in the root of the project or in" " the root directory of an account.",
284	)
285
286	if not list(path.glob("*.tf")):	×
287	raise ExitError(1, "This command can only run at [bold]layer[/bold] level.")	×
288
289	def check_for_cluster_layer(self, path: Path = None):	1✔
290	path = path or self.cwd	1✔
291	self.check_for_layer_location(path)	1✔
292	# assuming the "cluster" layer will contain the expected EKS outputs
293	if path.parts[-1] != "cluster":	1✔
294	raise ExitError(1, "This command can only run at the [bold]cluster layer[/bold].")	1✔
295
296
297	def get_project_root_or_current_dir_path() -> Path:	1✔
298	"""Returns the project root if detected, otherwise the current path"""
299	try:	1✔
300	root = Path(get_root_path())	1✔
301	except (NotARepositoryError, TypeError):	1✔
302	root = Path.cwd()	1✔
303
304	return root	1✔

binbashar / leverage / 17239897634

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