Format

Author: angelofenoglio

leverage/modules/auth.py

@@ -131,6 +131,7 @@ def authenticate(command):
         def some_command(tf: TFRunner, paths: PathsHandler, args):
             # command logic
     """
+
     @wraps(command)
     def new_command(*args, **kwargs):
         ctx = click.get_current_context()
@@ -251,6 +252,7 @@ def refresh_layer_credentials(paths: PathsHandler):
         )
         logger.info(f"Credentials for {account_name} account written successfully.")
 
+
 def refresh_layer_credentials_mfa(paths: PathsHandler):
     tf_profile, raw_profiles = get_profiles(paths)
     config_updater = ConfigUpdater()
@@ -271,7 +273,11 @@ def refresh_layer_credentials_mfa(paths: PathsHandler):
         layer_profile = tf_profile if raw_profile in ("${var.profile}", "each.value.profile") else None
 
         # replace variables with their corresponding values
-        profile_name = raw_profile.replace("${var.profile}", tf_profile).replace("${var.project}", paths.project).replace("each.value.profile", tf_profile)
+        profile_name = (
+            raw_profile.replace("${var.profile}", tf_profile)
+            .replace("${var.project}", paths.project)
+            .replace("each.value.profile", tf_profile)
+        )
 
         # if layer_profile wasn't set, use profile_name
         if layer_profile is None:
@@ -286,14 +292,17 @@ def refresh_layer_credentials_mfa(paths: PathsHandler):
             raise ExitError(
                 40,
                 f"Credentials for profile {profile_name} have not been properly configured. Please check your configuration.\n"
-                f"Check the following link for possible solutions: https://leverage.binbash.co/user-guide/troubleshooting/credentials/")
-        
+                f"Check the following link for possible solutions: https://leverage.binbash.co/user-guide/troubleshooting/credentials/",
+            )
+
         cache_file = paths.aws_cache_dir / profile_name
         if cache_file.exists():
             logger.debug(f"Found cached credentials in {cache_file}.")
             cached_credentials = json.loads(cache_file.read_text())
-            
-            expiration = datetime.strptime(cached_credentials.get("Expiration"), "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=tzutc())
+
+            expiration = datetime.strptime(cached_credentials.get("Expiration"), "%Y-%m-%dT%H:%M:%SZ").replace(
+                tzinfo=tzutc()
+            )
             renewal = datetime.now(tzutc()) + timedelta(seconds=(30 * 60))
             if renewal < expiration:
                 logger.info("Using cached credentials.")
@@ -302,10 +311,7 @@ def refresh_layer_credentials_mfa(paths: PathsHandler):
         else:
             logger.debug("No cached credentials found.")
 
-        client_session = boto3.Session(
-            botocore_session=session,
-            profile_name=source_profile
-        )
+        client_session = boto3.Session(botocore_session=session, profile_name=source_profile)
         client = client_session.client("sts")
         credentials = None
         for _ in range(3):
@@ -315,7 +321,9 @@ def refresh_layer_credentials_mfa(paths: PathsHandler):
                 raise ExitError(1, "Aborted by user.")
 
             try:
-                logger.debug(f"Assuming role {role_arn} for {profile_name} profile with serial {mfa_serial} and token code {mfa_token_code}")
+                logger.debug(
+                    f"Assuming role {role_arn} for {profile_name} profile with serial {mfa_serial} and token code {mfa_token_code}"
+                )
                 credentials = client.assume_role(
                     RoleArn=role_arn,
                     SourceIdentity=source_profile,
@@ -342,15 +350,15 @@ def refresh_layer_credentials_mfa(paths: PathsHandler):
                 elif error.response["Error"]["Code"] == "ExpiredToken":
                     logger.error("Token has expired. Please try again.")
                     continue
-                elif error.response["Error"]["Code"] == "ValidationError" and "Invalid length for parameter TokenCode" in error.response["Error"]["Message"]:
+                elif (
+                    error.response["Error"]["Code"] == "ValidationError"
+                    and "Invalid length for parameter TokenCode" in error.response["Error"]["Message"]
+                ):
                     logger.error("Invalid token length, it must be 6 digits long. Please try again.")
                     continue
                 elif "An error occurred" in error.response["Error"]["Message"]:
-                    raise ExitError(
-                        50,
-                        f"Error assuming role: {error}"
-                    )
-        
+                    raise ExitError(50, f"Error assuming role: {error}")
+
         if credentials is None:
             raise ExitError(60, "Failed to get credentials after 3 attempts. Please try again later.")
 

leverage/modules/kubectl.py

@@ -53,12 +53,10 @@ def kubectl(context, state, args):
     )
 
     authenticate = pass_paths(lambda paths: perform_authentication(paths))
-    _handle_subcommand(
-        context=context, runner=state.runner, args=args, pre_invocation_callback=authenticate
-    )
+    _handle_subcommand(context=context, runner=state.runner, args=args, pre_invocation_callback=authenticate)
 
 
-def _configure(environment: dict, ci: ClusterInfo = None,layer_path: Path = None):
+def _configure(environment: dict, ci: ClusterInfo = None, layer_path: Path = None):
     """
     Add the given EKS cluster configuration to the .kube/ files.
     """

leverage/modules/runner.py

@@ -95,15 +95,23 @@ def run(
         logger.debug(f"Working directory: {working_dir or Path.cwd()}")
         logger.debug(f"Additional environment variables: {merged_env_vars}")
 
-        process = subprocess.run(command, env=env, cwd=working_dir, capture_output=not interactive, text=not interactive)
+        process = subprocess.run(
+            command, env=env, cwd=working_dir, capture_output=not interactive, text=not interactive
+        )
 
         if raises and not interactive and process.returncode:
-            raise ExitError(process.returncode,  f"Command execution failed: {process.stderr.strip()}")
+            raise ExitError(process.returncode, f"Command execution failed: {process.stderr.strip()}")
 
-        return process.returncode if interactive else (process.returncode, process.stdout.strip(), process.stderr.strip())
+        return (
+            process.returncode if interactive else (process.returncode, process.stdout.strip(), process.stderr.strip())
+        )
 
     def exec(
-        self, *args: str, env_vars: Optional[Dict[str, str]] = None, working_dir: Optional[Path] = None, raises: bool = True
+        self,
+        *args: str,
+        env_vars: Optional[Dict[str, str]] = None,
+        working_dir: Optional[Path] = None,
+        raises: bool = True,
     ) -> Tuple[int, str, str]:
         """
         Execute command with the binary in non-interactive mode (captures output).

leverage/modules/tfrunner.py

@@ -70,7 +70,13 @@ def run(
         """
         return super().run(*args, env_vars=env_vars, working_dir=working_dir, raises=raises)
 
-    def exec(self, *args: str, env_vars: Optional[Dict[str, str]] = None, working_dir: Optional[Path] = None, raises: bool = False):
+    def exec(
+        self,
+        *args: str,
+        env_vars: Optional[Dict[str, str]] = None,
+        working_dir: Optional[Path] = None,
+        raises: bool = False,
+    ):
         """
         Execute the Terraform/OpenTofu binary in non-interactive mode (captures output).
 

leverage/modules/utils.py

@@ -36,10 +36,12 @@ def _handle_subcommand(
             pre_invocation_callback()
         exit_code = runner.run(*args)
         raise Exit(exit_code)
-    
+
     subcommand = context.command.commands.get(subcommand)
     # Check that the subcommand arguments are valid
-    subcommand.make_context(info_name=subcommand.name, args=list(args)[args.index(subcommand.name) + 1:], parent=context)
+    subcommand.make_context(
+        info_name=subcommand.name, args=list(args)[args.index(subcommand.name) + 1 :], parent=context
+    )
     # Invoke wrapped command
     if not subcommand.params:
         context.invoke(subcommand)

leverage/path.py

@@ -129,7 +129,7 @@ def get_build_script_path(filename="build.py"):
         cur_path = cur_path.parent
 
 
-class PathsHandler: # TODO: Turn this class into a something that represents a leverage project
+class PathsHandler:  # TODO: Turn this class into a something that represents a leverage project
     COMMON_TF_VARS = "common.tfvars"
     ACCOUNT_TF_VARS = "account.tfvars"
     BACKEND_TF_VARS = "backend.tfvars"
@@ -200,7 +200,7 @@ def aws_config_file(self):
     @property
     def aws_credentials_file(self):
         return self.aws_credentials_dir / "credentials"
-    
+
     @property
     def aws_cache_dir(self):
         return self.aws_credentials_dir / "cache"

tests/conftest.py

@@ -121,10 +121,7 @@ def leverage_project(tmp_path):
     # Create account/config
     account_config = account_dir / "config"
     account_config.mkdir()
-    (account_config / "account.tfvars").write_text(
-        'environment = "account"\n'
-        'sso_role = "test-sso-role"\n'
-    )
+    (account_config / "account.tfvars").write_text('environment = "account"\n' 'sso_role = "test-sso-role"\n')
     (account_config / "backend.tfvars").write_text(
         'profile = "bb-account-profile"\n'
         'bucket = "bb-account-terraform-backend"\n'
@@ -153,11 +150,11 @@ def leverage_project(tmp_path):
     security_base.mkdir()
     (security_base / "security-base.tf").write_text("# Security base configuration\n")
     (security_base / "config.tf").write_text(
-        'terraform {\n'
+        "terraform {\n"
         '  backend "s3" {\n'
         '    key = "account/us-east-1/security-base/terraform.tfstate"\n'
-        '  }\n'
-        '}\n'
+        "  }\n"
+        "}\n"
     )
     (security_base / "backend.tfvars").write_text(
         'profile = "bb-account-profile"\n'
@@ -168,6 +165,7 @@ def leverage_project(tmp_path):
 
     return root
 
+
 @pytest.fixture
 def leverage_runner(monkeypatch):
     """
@@ -218,6 +216,7 @@ def runner(leverage_directory):
 
     return runner
 
+
 @pytest.fixture
 def leverage_context(leverage_project, monkeypatch):
     def context(verbose=True, build_script_name="build.py"):
@@ -238,4 +237,4 @@ def context(verbose=True, build_script_name="build.py"):
 
         return click.Context(command=click.Command("leverage"), obj=state)
 
-    return context
+    return context

tests/test_modules/test_kubectl.py

@@ -7,6 +7,7 @@
 from leverage import leverage
 from leverage.modules.kubectl import _scan_clusters, ClusterInfo
 
+
 def test_scan_clusters():
     """
     Test that we can find valid metadata.yaml presents in the down the path of the filesystem tree where we are staying.
@@ -38,7 +39,9 @@ def test_discover(leverage_project):
     }
     cli_runner = CliRunner()
     with cli_runner.isolated_filesystem(leverage_project) as leverage_project_folder:
-        with patch("leverage.modules.kubectl._scan_clusters", return_value=[(leverage_project_folder, mocked_cluster_data)]) as mkd_scan_clusters:
+        with patch(
+            "leverage.modules.kubectl._scan_clusters", return_value=[(leverage_project_folder, mocked_cluster_data)]
+        ) as mkd_scan_clusters:
             with patch("simple_term_menu.TerminalMenu") as mkd_show:
                 mkd_show.return_value.show.return_value = 0  # simulate choosing the first result
                 with patch("leverage.modules.kubectl._configure") as mkd_configure:

tests/test_modules/test_tf.py

@@ -46,7 +46,10 @@ def test_init_with_args(leverage_project, leverage_runner):
 
         assert mocked_run.call_args_list[0][0][0] == "init"
         assert mocked_run.call_args_list[0][0][1] == "-migrate-state"
-        assert mocked_run.call_args_list[0][0][2] == f"-backend-config={leverage_project / 'account' / 'config' / 'backend.tfvars'}"
+        assert (
+            mocked_run.call_args_list[0][0][2]
+            == f"-backend-config={leverage_project / 'account' / 'config' / 'backend.tfvars'}"
+        )
 
 
 @pytest.mark.parametrize(