• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

binbashar / leverage / 17239897634

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

push

github

web-flow
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

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
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✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc