11999658091

Committed 24 Nov 2024 09:16PM UTC coverage: 60.419% (+0.4%) from 60.037%

Build # 11999658091

Build Type

Pull #281

github

Committed by

Franr

Commit Message

removing assert

Pull Request Pull Request #281: [BV-599] Kubeconfig helper

Run Details

196 of 460 branches covered (42.61%)

Branch coverage included in aggregate %.

52 of 73 new or added lines in 4 files covered. (71.23%)

47 existing lines in 5 files now uncovered.

2518 of 4032 relevant lines covered (62.45%)

0.62 hits per line

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

71.1

/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 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 Terraform command can be run here.")

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

binbashar / leverage / 11999658091

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