From e975819244f7e620607de3cfc33abc5977205c46 Mon Sep 17 00:00:00 2001 From: Diego OJ Date: Sun, 30 Aug 2026 16:52:47 -0300 Subject: [PATCH 1/2] Fix | Restore the unit test suite The unit test workflow has been failing on master since #316, across three consecutive merges. This restores it on the whole 3.9 - 3.13 matrix. Reconcile test_credentials.py with the dockerless design: #316 rewrote leverage/modules/credentials.py but left its tests untouched, so the module failed to import and pytest aborted during collection, which masked every other failure in the run. The tests now inject the runner, paths and config through the click context, as `pass_runner`, `pass_paths` and `pass_state` expect, and account for `Runner.exec` returning a (exit code, stdout, stderr) triple and for the `-mfa` profile suffix that `refresh_layer_credentials_mfa` looks up. Fix _update_account_ids corrupting common.tfvars: #316 replaced the hcledit call with a non-greedy regex that stops at the first closing brace. On a nested `accounts` block, which is what the reference architecture ships, it replaced only the first account, left the remaining ones orphaned outside the block and produced invalid HCL with unbalanced braces. Replacement now scans for the matching brace, and is anchored so `external_accounts` is no longer a candidate match. Fix mock target resolution on Python 3.9: `leverage/modules/__init__.py` re-exports the click Groups, so `leverage.modules.` resolves to the Group rather than to the module. From 3.10 on mock resolves these targets through `pkgutil.resolve_name` and finds the module anyway, but on 3.9 it walks attributes and finds the Group, raising AttributeError during setup. The affected targets are now patched by object, which behaves the same on every supported version. Update test_tf.py for the tfvars injected into init, added deliberately in #316 for ref-arch v2. The tfvars are discovered by globbing, so their order depends on the filesystem and is asserted as a set. Use a raw string for the regex in test_path.py, silencing a SyntaxWarning that becomes a SyntaxError in a future Python version. Co-Authored-By: Claude Opus 5 (1M context) --- leverage/modules/credentials.py | 68 +++- tests/test_modules/test_auth.py | 24 +- tests/test_modules/test_credentials.py | 417 ++++++++++++++++--------- tests/test_modules/test_kubectl.py | 12 +- tests/test_modules/test_tf.py | 40 ++- tests/test_path.py | 2 +- 6 files changed, 378 insertions(+), 185 deletions(-) diff --git a/leverage/modules/credentials.py b/leverage/modules/credentials.py index 4a0d7d7..0de0dfd 100644 --- a/leverage/modules/credentials.py +++ b/leverage/modules/credentials.py @@ -620,6 +620,70 @@ def configure_accounts_profiles( configure_profile(profile_identifier, profile_values) +def _find_matching_brace(content: str, start: int): + """Find the position of the brace closing the one opened at `start`. + + Braces appearing inside double quoted strings are ignored. + + Args: + content (str): Text to scan. + start (int): Position of the opening brace. + + Returns: + int: Position of the matching closing brace, or None if it is unbalanced. + """ + depth = 0 + in_string = False + position = start + + while position < len(content): + char = content[position] + + if in_string: + if char == "\\": + position += 1 + elif char == '"': + in_string = False + elif char == '"': + in_string = True + elif char == "{": + depth += 1 + elif char == "}": + depth -= 1 + if not depth: + return position + + position += 1 + + return None + + +def _replace_hcl_attribute(content: str, attribute: str, value: str): + """Replace the value of a brace delimited HCL attribute, honoring nested blocks. + + A non greedy regex cannot be used for this: it stops at the first closing brace, which for a + nested block leaves the remaining entries orphaned and the file with unbalanced braces. + + Args: + content (str): Full text of the HCL file. + attribute (str): Name of the attribute to replace, e.g. `accounts`. + value (str): New value for the attribute, enclosing braces included. + + Returns: + str: Content with the attribute replaced, unchanged if the attribute was not found. + """ + attribute_definition = re.search(rf"^{re.escape(attribute)}\s*=\s*\{{", content, flags=re.MULTILINE) + if attribute_definition is None: + return content + + opening_brace = attribute_definition.end() - 1 + closing_brace = _find_matching_brace(content, opening_brace) + if closing_brace is None: + return content + + return f"{content[:attribute_definition.start()]}{attribute} = {value}{content[closing_brace + 1:]}" + + @pass_paths def _update_account_ids(paths: PathsHandler, config: dict): """Update accounts ids in global configuration file. @@ -654,9 +718,7 @@ def _update_account_ids(paths: PathsHandler, config: dict): accs = f"{{{accs}\n}}" common_tfvars = paths.common_tfvars.read_text() - common_tfvars = re.sub( - r"accounts\s*=\s*\{.*?\}(?=\s*(?:\n|$))", f"accounts = {accs}", common_tfvars, flags=re.DOTALL - ) + common_tfvars = _replace_hcl_attribute(common_tfvars, "accounts", accs) paths.common_tfvars.write_text(common_tfvars) diff --git a/tests/test_modules/test_auth.py b/tests/test_modules/test_auth.py index 81a7526..abb994b 100644 --- a/tests/test_modules/test_auth.py +++ b/tests/test_modules/test_auth.py @@ -1,4 +1,5 @@ from collections import namedtuple +from importlib import import_module from pathlib import PosixPath from unittest import mock from unittest.mock import Mock, MagicMock @@ -18,6 +19,11 @@ ) from leverage.modules.aws import get_account_roles, add_sso_profile, configure_sso_profiles +# `leverage.modules.aws` resolves to the click Group of the same name, since it is re-exported +# in `leverage/modules/__init__.py`. Patching by string would target that Group instead of the +# module, so the module itself is imported and patched by object. +aws_module = import_module("leverage.modules.aws") + @pytest.fixture def paths(with_click_context, propagate_logs): @@ -47,8 +53,8 @@ def mock_sso_token(): Patches both bindings (auth.py defines it, aws.py imports it) so any caller in either module hits the same fake. """ - with mock.patch("leverage.modules.auth.get_sso_access_token", return_value="testing-token") as m, mock.patch( - "leverage.modules.aws.get_sso_access_token", return_value="testing-token" + with mock.patch("leverage.modules.auth.get_sso_access_token", return_value="testing-token") as m, mock.patch.object( + aws_module, "get_sso_access_token", return_value="testing-token" ): yield m @@ -106,9 +112,9 @@ def test_add_sso_profile(): @mock.patch("boto3.client") def test_configure_sso_profiles(mocked_boto, paths, mock_sso_token): - with mock.patch("leverage.modules.aws.ConfigUpdater.__new__", return_value=mocked_updater): - with mock.patch("leverage.modules.aws.get_account_roles", return_value=ACC_ROLES): - with mock.patch("leverage.modules.aws.add_sso_profile") as mocked_add_profile: + with mock.patch.object(aws_module.ConfigUpdater, "__new__", return_value=mocked_updater): + with mock.patch.object(aws_module, "get_account_roles", return_value=ACC_ROLES): + with mock.patch.object(aws_module, "add_sso_profile") as mocked_add_profile: configure_sso_profiles(paths) # 2 profiles were added @@ -254,7 +260,7 @@ def open_side_effect(name, *args, **kwargs): @mock.patch("leverage.modules.auth.get_profiles", new=Mock(return_value=("test-first-devops", ["test-first-profile"]))) @mock.patch("leverage.modules.auth.get_or_create_section", new=Mock()) -@mock.patch("leverage.modules.aws.ConfigUpdater.update_file", new=Mock()) +@mock.patch.object(aws_module.ConfigUpdater, "update_file", new=Mock()) @mock.patch("pathlib.Path.touch", new=Mock()) @mock.patch("boto3.client", return_value=b3_client) @mock.patch("configupdater.parser.open", side_effect=open_side_effect) @@ -269,7 +275,7 @@ def test_refresh_layer_credentials_first_time(mock_open, mock_boto, paths, mock_ @mock.patch("leverage.modules.auth.get_profiles", new=Mock(return_value=("test-valid-devops", ["test-valid-profile"]))) @mock.patch("leverage.modules.auth.get_or_create_section", new=Mock()) -@mock.patch("leverage.modules.aws.ConfigUpdater.update_file", new=Mock()) +@mock.patch.object(aws_module.ConfigUpdater, "update_file", new=Mock()) @mock.patch("time.time", new=Mock(return_value=NOW_EPOCH)) @mock.patch("boto3.client", return_value=b3_client) @mock.patch("configupdater.parser.open", side_effect=open_side_effect) @@ -600,7 +606,7 @@ def test_refresh_all_accounts_credentials_integration(tmp_path, paths, mock_sso_ def test_aws_sso_refresh_invokes_refresh_all_accounts(leverage_project, leverage_runner): """`leverage aws sso refresh` reaches refresh_all_accounts_credentials with force_refresh=False.""" with leverage_runner(leverage_project) as runner: - with mock.patch("leverage.modules.aws.refresh_all_accounts_credentials") as mock_refresh: + with mock.patch.object(aws_module, "refresh_all_accounts_credentials") as mock_refresh: result = runner.invoke(leverage, ["aws", "sso", "refresh"]) assert result.exit_code == 0, result.output + (str(result.exception) if result.exception else "") @@ -611,7 +617,7 @@ def test_aws_sso_refresh_invokes_refresh_all_accounts(leverage_project, leverage def test_aws_sso_refresh_force_invokes_refresh_all_accounts(leverage_project, leverage_runner): """`leverage aws sso refresh --force` forwards force_refresh=True.""" with leverage_runner(leverage_project) as runner: - with mock.patch("leverage.modules.aws.refresh_all_accounts_credentials") as mock_refresh: + with mock.patch.object(aws_module, "refresh_all_accounts_credentials") as mock_refresh: result = runner.invoke(leverage, ["aws", "sso", "refresh", "--force"]) assert result.exit_code == 0, result.output + (str(result.exception) if result.exception else "") diff --git a/tests/test_modules/test_credentials.py b/tests/test_modules/test_credentials.py index a73e719..1933315 100644 --- a/tests/test_modules/test_credentials.py +++ b/tests/test_modules/test_credentials.py @@ -1,9 +1,13 @@ +from contextlib import contextmanager +from importlib import import_module from pathlib import Path from unittest import mock from unittest.mock import Mock +import click import pytest +from leverage._internals import State from leverage._utils import ExitError from leverage.modules.credentials import ( _load_configs_for_credentials, @@ -11,8 +15,7 @@ _extract_credentials, _get_mfa_serial, _get_organization_accounts, - _profile_is_configured, - _backup_file, + _replace_hcl_attribute, configure_credentials, _credentials_are_valid, _get_management_account_id, @@ -20,125 +23,139 @@ _update_account_ids, ) -mocked_aws_cli = Mock() +# `leverage.modules.credentials` resolves to the click Group of the same name, since it is +# re-exported in `leverage/modules/__init__.py`. Patching by string would target that Group +# instead of the module, so the module itself is imported and patched by object. +credentials_module = import_module("leverage.modules.credentials") -@mock.patch( - "leverage.modules.credentials._load_project_yaml", - Mock( - return_value={ - "short_name": "test", - "region": "us-test-1", +@contextmanager +def cli_context(runner=None, paths=None, config=None, verbose=False): + """Build a Leverage click context holding the given runner, paths and configuration. + + The credentials module gets all three injected through the `pass_runner`, `pass_paths` + and `pass_state` decorators, so they must be set on the context state object. + """ + state = State() + state.verbosity = verbose + state.runner = runner + state.paths = paths + state.config = config + + with click.Context(command=click.Command("leverage"), obj=state): + yield + + +def awscli_returning(exit_code, output): + """AWS cli runner double whose `exec` returns the given exit code and output.""" + return Mock(exec=Mock(return_value=(exit_code, output, ""))) + + +PROJECT_YAML = { + "short_name": "test", + "region": "us-test-1", + "organization": {"accounts": [{"name": "acc2"}]}, +} + +ENV_CONFIG = {"PROJECT": "test", "MFA_ENABLED": "true"} + +COMMON_CONF = { + "project_long": "test-prjt", + "region_secondary": "us-test-2", + "accounts": {"acc1": {"email": "test@test.com", "id": "123456"}}, +} + + +@mock.patch.object(credentials_module, "_load_project_yaml", Mock(return_value=PROJECT_YAML)) +def test_load_configs_for_credentials(): + """ + Test that the values needed to configure the credentials are gathered from the project + configuration file, the build.env config and the tf common configuration. + """ + with cli_context(paths=Mock(common_conf=COMMON_CONF), config=ENV_CONFIG): + assert _load_configs_for_credentials() == { + "mfa_enabled": "true", "organization": { "accounts": [ - {"name": "acc2"}, - ] - }, - } - ), -) -@mock.patch( - "leverage.modules.credentials.AWSCLI", - Mock( - env_conf={ - "PROJECT": "test", - "MFA_ENABLED": "true", - }, - paths=Mock( - common_conf={ - "project_long": "test-prjt", - "region_secondary": "us-test-2", - "accounts": { - "acc1": { + { "email": "test@test.com", "id": "123456", - } - }, + "name": "acc1", + }, + { + "name": "acc2", + }, + ] }, - ), - ), -) -def test_load_configs_for_credentials(with_click_context): - assert _load_configs_for_credentials() == { - "mfa_enabled": "true", - "organization": { - "accounts": [ - { - "email": "test@test.com", - "id": "123456", - "name": "acc1", - }, - { - "name": "acc2", - }, - ] - }, - "primary_region": "us-test-1", - "project_name": "test-prjt", - "secondary_region": "us-test-2", - "short_name": "test", - } + "primary_region": "us-test-1", + "project_name": "test-prjt", + "secondary_region": "us-test-2", + "short_name": "test", + } -@mock.patch("leverage.modules.credentials._get_mfa_serial", new=Mock(return_value="mfa123")) -@mock.patch("leverage.modules.credentials._backup_file") -def test_configure_accounts_profiles(mocked_backup, muted_click_context): +@mock.patch.object(credentials_module, "_get_mfa_serial", new=Mock(return_value="mfa123")) +@mock.patch.object(credentials_module.shutil, "copy") +def test_configure_accounts_profiles(mocked_copy): """ Test that the expected jsons for the aws credentials are generated as expected. No-mfa case. """ - with mock.patch("leverage.modules.credentials.configure_profile") as mocked_config: - configure_accounts_profiles( - "test-management", - "us-test-1", - {"acc1": "12345", "out-of-project-acc": "67890"}, - [{"name": "acc1"}], - fetch_mfa_device=False, - ) + paths = Mock() + with cli_context(paths=paths): + with mock.patch.object(credentials_module, "configure_profile") as mocked_config: + configure_accounts_profiles( + "test-management", + "us-test-1", + {"acc1": "12345", "out-of-project-acc": "67890"}, + [{"name": "acc1"}], + fetch_mfa_device=False, + ) - # make sure we did a backup with the old credentials - assert mocked_backup.assert_called_once + # make sure we did a backup of the previous account profiles + mocked_copy.assert_called_once_with(paths.aws_config_file, paths.aws_config_file.with_suffix(".bkp")) # only 1 call since "out-of-project-acc" should be avoided assert mocked_config.call_count == 1 - assert mocked_config.call_args_list[0][0][0] == "test-acc1-oaar" + assert mocked_config.call_args_list[0][0][0] == "test-acc1-oaar-mfa" expected = { "output": "json", "region": "us-test-1", - "role_arn": f"arn:aws:iam::12345:role/OrganizationAccountAccessRole", + "role_arn": "arn:aws:iam::12345:role/OrganizationAccountAccessRole", "source_profile": "test-management", } assert mocked_config.call_args_list[0][0][1] == expected -@pytest.mark.parametrize("mfa_device", [False, True]) -@mock.patch("leverage.modules.credentials._get_mfa_serial", new=Mock(return_value="mfa123")) -@mock.patch("leverage.modules.credentials._backup_file") -def test_configure_accounts_profiles_mfa(mocked_backup, mfa_device, muted_click_context): +@mock.patch.object(credentials_module, "_get_mfa_serial", new=Mock(return_value="mfa123")) +@mock.patch.object(credentials_module.shutil, "copy") +def test_configure_accounts_profiles_mfa(mocked_copy): """ Test that the expected jsons for the aws credentials are generated as expected. Mfa case. """ - with mock.patch("leverage.modules.credentials.configure_profile") as mocked_config: - configure_accounts_profiles( - "test-management", - "us-test-1", - {"acc1": "12345", "out-of-project-acc": "67890"}, - [{"name": "acc1"}], - fetch_mfa_device=True, - ) + paths = Mock() + with cli_context(paths=paths): + with mock.patch.object(credentials_module, "configure_profile") as mocked_config: + configure_accounts_profiles( + "test-management", + "us-test-1", + {"acc1": "12345", "out-of-project-acc": "67890"}, + [{"name": "acc1"}], + fetch_mfa_device=True, + ) - # make sure we did a backup with the old credentials - assert mocked_backup.assert_called_once + # make sure we did a backup of the previous account profiles + mocked_copy.assert_called_once_with(paths.aws_config_file, paths.aws_config_file.with_suffix(".bkp")) # only 1 call since "out-of-project-acc" should be avoided assert mocked_config.call_count == 1 - assert mocked_config.call_args_list[0][0][0] == "test-acc1-oaar" + assert mocked_config.call_args_list[0][0][0] == "test-acc1-oaar-mfa" expected = { "output": "json", "region": "us-test-1", - "role_arn": f"arn:aws:iam::12345:role/OrganizationAccountAccessRole", + "role_arn": "arn:aws:iam::12345:role/OrganizationAccountAccessRole", "source_profile": "test-management", "mfa_serial": "mfa123", } @@ -146,13 +163,14 @@ def test_configure_accounts_profiles_mfa(mocked_backup, mfa_device, muted_click_ assert mocked_config.call_args_list[0][0][1] == expected -@mock.patch("leverage.modules.credentials._get_mfa_serial", new=Mock(return_value="")) -def test_configure_accounts_profiles_mfa_error(muted_click_context): +@mock.patch.object(credentials_module, "_get_mfa_serial", new=Mock(return_value="")) +def test_configure_accounts_profiles_mfa_error(): """ Test that if we fail to fetch the MFA serial number, user get a proper error. """ - with pytest.raises(ExitError, match="No MFA device found for user."): - configure_accounts_profiles("test-management", "us-test-1", {}, [], True) + with cli_context(paths=Mock()): + with pytest.raises(ExitError, match="No MFA device found for user."): + configure_accounts_profiles("test-management", "us-test-1", {}, [], True) @mock.patch( @@ -177,8 +195,8 @@ def test_get_organization_accounts(): """ Test that the list of accounts of an organization are queried and returned in a {acc name: acc id} dict. """ - mocked_aws_cli.exec = Mock(return_value=(0, '{"Accounts": [{"Name": "test-acc1", "Id": "12345"}]}')) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(0, '{"Accounts": [{"Name": "test-acc1", "Id": "12345"}]}') + with cli_context(runner=awscli): assert _get_organization_accounts("foo", "bar") == {"test-acc1": "12345"} @@ -186,8 +204,8 @@ def test_get_organization_accounts_error(): """ Test that, if getting the list of accounts fails for some reason, we return an empty dict. """ - mocked_aws_cli.exec = Mock(return_value=(1, "BAD")) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(1, "BAD") + with cli_context(runner=awscli): assert _get_organization_accounts("foo", "bar") == {} @@ -195,19 +213,17 @@ def test_get_mfa_serial(): """ Test that we fetch the mfa devices from the profile and return the serial number of the first one that is valid. """ - mocked_aws_cli.exec = Mock( - return_value=(0, '{"MFADevices": [{"SerialNumber": "arn:aws:iam::123456789012:mfa/testuser"}]}') - ) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(0, '{"MFADevices": [{"SerialNumber": "arn:aws:iam::123456789012:mfa/testuser"}]}') + with cli_context(runner=awscli): assert _get_mfa_serial("foo") == "arn:aws:iam::123456789012:mfa/testuser" -def test_get_mfa_serial_error(muted_click_context): +def test_get_mfa_serial_error(): """ Test that, if fetching mfa devices fails, we return a user-friendly error. """ - mocked_aws_cli.exec = Mock(return_value=(1, "BAD")) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(1, "BAD") + with cli_context(runner=awscli): with pytest.raises(ExitError, match="AWS CLI error: BAD"): _get_mfa_serial("foo") @@ -216,87 +232,184 @@ def test_credentials_are_valid(): """ Test that AWS credentials for the current profile are valid. """ - mocked_aws_cli.exec = Mock(return_value=(0, "OK")) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(0, "OK") + with cli_context(runner=awscli): assert _credentials_are_valid("foo") +def test_credentials_are_not_valid(): + """ + Test that an invalid security token is reported as invalid credentials. + """ + awscli = awscli_returning( + 255, + "An error occurred (InvalidClientTokenId) when calling the GetCallerIdentity operation:" + " The security token included in the request is invalid.", + ) + with cli_context(runner=awscli): + assert not _credentials_are_valid("foo") + + def test_get_management_account_id(): """ Test that we can get the account id from the current profile. """ - mocked_aws_cli.exec = Mock(return_value=(0, '{"Account": "123456789012"}')) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(0, '{"Account": "123456789012"}') + with cli_context(runner=awscli): assert _get_management_account_id("foo") == "123456789012" -def test_get_management_account_id_error(with_click_context): +def test_get_management_account_id_error(): """ Test that we return a user-friendly error if getting the account id of a profile fails. """ - mocked_aws_cli.exec = Mock(return_value=(1, "BAD")) - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): + awscli = awscli_returning(1, "BAD") + with cli_context(runner=awscli): with pytest.raises(ExitError, match="AWS CLI error: BAD"): _get_management_account_id("foo") -def test_configure_credentials(with_click_context, propagate_logs, caplog): +def test_configure_profile(): + """ + Test that every value of a profile is set through the AWS cli. + """ + awscli = awscli_returning(0, "") + with cli_context(runner=awscli): + configure_profile("test-acc1-oaar-mfa", {"region": "us-test-1", "output": "json"}) + + assert awscli.exec.call_args_list == [ + mock.call("configure", "set", "region", "us-test-1", "--profile", "test-acc1-oaar-mfa"), + mock.call("configure", "set", "output", "json", "--profile", "test-acc1-oaar-mfa"), + ] + + +@mock.patch.object(credentials_module, "_ask_for_credentials", new=Mock(return_value=("foo", "bar"))) +@mock.patch.object(credentials_module.shutil, "copy") +def test_configure_credentials(mocked_copy, propagate_logs, caplog): """ Test that the aws credentials for the profile are set and the backup feature is called. """ - mocked_aws_cli.exec = Mock(return_value=(0, "")) - with mock.patch("leverage.modules.credentials._backup_file"): - with mock.patch("leverage.modules.credentials._ask_for_credentials", new=Mock(return_value=("foo", "bar"))): - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): - configure_credentials("foo", "manual", make_backup=True) + paths = Mock() + with cli_context(runner=awscli_returning(0, ""), paths=paths, verbose=True): + configure_credentials("foo", "manual", make_backup=True) assert caplog.messages[0] == "Backing up credentials file." + mocked_copy.assert_called_once_with(paths.aws_credentials_file, paths.aws_credentials_file.with_suffix(".bkp")) -def test_configure_credentials_error(with_click_context): +@mock.patch.object(credentials_module, "_extract_credentials", new=Mock(return_value=("foo", "bar"))) +def test_configure_credentials_error(): """ Test that, if settings the credentials for a profile fails, we return a user-friendly error. """ - mocked_aws_cli.exec = Mock(return_value=(1, "BROKEN")) - with mock.patch("leverage.modules.credentials._extract_credentials", new=Mock(return_value=("foo", "bar"))): - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): - with pytest.raises(ExitError, match="AWS CLI error: BROKEN"): - configure_credentials("foo", "/.aws/creds") - - -def test_update_account_ids(with_click_context, propagate_logs): - """ - Test that account ids are updated in global configuration files. - """ - mocked_aws_cli.system_exec = Mock() - with mock.patch("leverage.modules.credentials.PROJECT_COMMON_TFVARS"): - with mock.patch("leverage.modules.credentials.AWSCLI", mocked_aws_cli): - _update_account_ids( - { - "project_name": "test", - "organization": { - "accounts": [ - { - "name": "acc1", - "email": "acc@test.com", - "id": "12345", - } - ] - }, - } - ) + with cli_context(runner=awscli_returning(1, "BROKEN"), paths=Mock()): + with pytest.raises(ExitError, match="AWS CLI error: BROKEN"): + configure_credentials("foo", "/.aws/creds") + + +def test_update_account_ids(tmp_path): + """ + Test that account ids are updated in global configuration files, replacing the whole + previous `accounts` block and leaving the rest of the file untouched. + """ + common_tfvars = tmp_path / "common.tfvars" + common_tfvars.write_text( + 'project = "bb"\n' + "\n" + "accounts = {\n" + " old = {\n" + ' email = "old@test.com",\n' + ' id = "00000"\n' + " },\n" + " other = {\n" + ' email = "other@test.com",\n' + ' id = "11111"\n' + " }\n" + "}\n" + "\n" + 'region_primary = "us-east-1"\n' + ) + + with cli_context(paths=Mock(common_tfvars=common_tfvars)): + _update_account_ids( + { + "project_name": "test", + "organization": { + "accounts": [ + { + "name": "acc1", + "email": "acc@test.com", + "id": "12345", + } + ] + }, + } + ) + + assert common_tfvars.read_text() == ( + 'project = "bb"\n' + "\n" + "accounts = {\n" + " acc1 = {\n" + ' email = "acc@test.com",\n' + ' id = "12345"\n' + " }\n" + "}\n" + "\n" + 'region_primary = "us-east-1"\n' + ) - assert ( - mocked_aws_cli.system_exec.call_args_list[0][0][0] - == 'hcledit -f /test/config/common.tfvars -u attribute set acc1_account_id "\\"12345\\""' + +def test_update_account_ids_without_common_tfvars(tmp_path): + """ + Test that nothing is attempted when the common tfvars file does not exist. + """ + with cli_context(paths=Mock(common_tfvars=tmp_path / "missing.tfvars")): + _update_account_ids({"organization": {"accounts": []}}) + + +def test_replace_hcl_attribute_honors_nested_blocks(): + """ + Test that the whole nested block is replaced. A non-greedy regex stops at the first closing + brace, orphaning the remaining entries and leaving the file with unbalanced braces. + """ + content = ( + "accounts = {\n" + " first = {\n" + ' id = "1"\n' + " },\n" + " second = {\n" + ' id = "2"\n' + " }\n" + "}\n" + "\n" + 'region = "us-east-1"\n' ) - assert ( - mocked_aws_cli.system_exec.call_args_list[1][0][0] - == """hcledit -f /test/config/common.tfvars -u attribute set accounts '{ - acc1 = { - email = \"acc@test.com\", - id = \"12345\" - } -}'""" + replaced = _replace_hcl_attribute(content, "accounts", '{\n only = {\n id = "3"\n }\n}') + + assert replaced == ("accounts = {\n" " only = {\n" ' id = "3"\n' " }\n" "}\n" "\n" 'region = "us-east-1"\n') + assert replaced.count("{") == replaced.count("}") + + +def test_replace_hcl_attribute_ignores_similarly_named_attributes(): + """ + Test that an attribute whose name ends with the target one is not replaced. + """ + content = ( + 'external_accounts = {\n drata = {\n id = "1"\n }\n}\n\naccounts = {\n old = {\n id = "2"\n }\n}\n' ) + + replaced = _replace_hcl_attribute(content, "accounts", '{\n new = {\n id = "3"\n }\n}') + + assert 'external_accounts = {\n drata = {\n id = "1"\n }\n}' in replaced + assert 'accounts = {\n new = {\n id = "3"\n }\n}' in replaced + + +def test_replace_hcl_attribute_missing_attribute(): + """ + Test that content without the attribute is returned unchanged. + """ + content = 'project = "bb"\n' + + assert _replace_hcl_attribute(content, "accounts", "{}") == content diff --git a/tests/test_modules/test_kubectl.py b/tests/test_modules/test_kubectl.py index 35ed597..9c0ea02 100644 --- a/tests/test_modules/test_kubectl.py +++ b/tests/test_modules/test_kubectl.py @@ -1,3 +1,4 @@ +from importlib import import_module from pathlib import Path, PosixPath from unittest import mock from unittest.mock import Mock, patch @@ -7,6 +8,11 @@ from leverage import leverage from leverage.modules.kubectl import _scan_clusters, ClusterInfo +# `leverage.modules.kubectl` resolves to the click Group of the same name, since it is +# re-exported in `leverage/modules/__init__.py`. Patching by string would target that Group +# instead of the module, so the module itself is imported and patched by object. +kubectl_module = import_module("leverage.modules.kubectl") + def test_scan_clusters(): """ @@ -39,12 +45,12 @@ 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)] + with patch.object( + kubectl_module, "_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: + with patch.object(kubectl_module, "_configure") as mkd_configure: cli_runner.invoke(leverage, ["kubectl", "discover"]) assert isinstance(mkd_configure.call_args_list[0][0][1], ClusterInfo) diff --git a/tests/test_modules/test_tf.py b/tests/test_modules/test_tf.py index d043f24..363603a 100644 --- a/tests/test_modules/test_tf.py +++ b/tests/test_modules/test_tf.py @@ -20,20 +20,26 @@ def test_init_arguments(leverage_project, leverage_runner, args): """ with leverage_runner(leverage_project) as runner: with patch("leverage.modules.tfrunner.TFRunner.run", return_value=0) as mocked_run: - result = runner.invoke(leverage, ["tf", "init", *args]) + runner.invoke(leverage, ["tf", "init", *args]) - # Check that init was called - assert mocked_run.call_args_list[0][0][0] == "init" + called_args = list(mocked_run.call_args_list[0][0]) - # Check that backend-config is included with the correct path - backend_config_path = str(leverage_project / "account" / "config" / "backend.tfvars") - backend_config_arg = f"-backend-config={backend_config_path}" + # Check that init was called + assert called_args[0] == "init" - # Build expected args: user args + backend-config - expected_args = list(args) + [backend_config_arg] - actual_args = list(mocked_run.call_args_list[0][0][1:]) + # The layer tfvars are injected before the user arguments. They are discovered by globbing + # the config directories, so their order depends on the filesystem and cannot be asserted. + assert {arg for arg in called_args if arg.startswith("-var-file=")} == { + f"-var-file={(leverage_project / 'config' / 'common.tfvars').as_posix()}", + f"-var-file={(leverage_project / 'account' / 'config' / 'account.tfvars').as_posix()}", + f"-var-file={(leverage_project / 'account' / 'config' / 'backend.tfvars').as_posix()}", + } - assert actual_args == expected_args + # Check that the user arguments are preserved and backend-config is appended last + backend_config_arg = f"-backend-config={leverage_project / 'account' / 'config' / 'backend.tfvars'}" + remaining_args = [arg for arg in called_args[1:] if not arg.startswith("-var-file=")] + + assert remaining_args == [*args, backend_config_arg] def test_init_with_args(leverage_project, leverage_runner): @@ -42,14 +48,14 @@ def test_init_with_args(leverage_project, leverage_runner): """ with leverage_runner(leverage_project) as runner: with patch("leverage.modules.tfrunner.TFRunner.run", return_value=0) as mocked_run: - result = runner.invoke(leverage, ["tf", "init", "-migrate-state"]) + runner.invoke(leverage, ["tf", "init", "-migrate-state"]) + + called_args = list(mocked_run.call_args_list[0][0]) - 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'}" - ) + # User arguments are placed after the layer tfvars and before the backend configuration + assert called_args[0] == "init" + assert called_args[-2] == "-migrate-state" + assert called_args[-1] == f"-backend-config={leverage_project / 'account' / 'config' / 'backend.tfvars'}" @pytest.mark.parametrize( diff --git a/tests/test_path.py b/tests/test_path.py index 84c1b3e..9eae5cd 100644 --- a/tests/test_path.py +++ b/tests/test_path.py @@ -115,7 +115,7 @@ def test_check_for_cluster_layer(muted_click_context, propagate_logs): """ paths = PathsHandler({"PROJECT": "test"}) with patch.object(paths, "check_for_layer_location"): # assume parent method is already tested - with pytest.raises(ExitError, match="This command can only run at the \[bold\]cluster layer\[/bold\]\."): + with pytest.raises(ExitError, match=r"This command can only run at the \[bold\]cluster layer\[/bold\]\."): paths.cwd = Path("/random") paths.check_for_cluster_layer() From 96f6fa2eb6849d53b31558068d961a62031b32f9 Mon Sep 17 00:00:00 2001 From: Diego OJ Date: Sun, 30 Aug 2026 17:24:40 -0300 Subject: [PATCH 2/2] Fix | Make the test suite independent of installed binaries The init tests reached TFRunner.run through the real constructor, so they needed tofu installed to get past binary discovery. Without it the command exited before run() was ever called and the assertions failed on an empty call list, which is what happens on the CI runners. test_discover had the same dependency on kubectl, which happens to be present on GitHub runners and so had been passing by luck. Binary discovery is skipped in both, since the execution itself is mocked. It keeps its own coverage in test_runner.py and test_tfrunner.py. Co-Authored-By: Claude Opus 5 (1M context) --- tests/conftest.py | 14 ++++++++++++++ tests/test_modules/test_kubectl.py | 4 +++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 559a198..51596d7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -179,6 +179,7 @@ def leverage_runner(monkeypatch): - get_root_path and get_working_path in both leverage.path and leverage.conf - Path.cwd() to return the working directory - check_sso_token and refresh_layer_credentials to skip authentication + - TFRunner binary discovery, so the suite does not require tofu to be installed Args: leverage_directory: Path to the root of the mock project @@ -186,6 +187,16 @@ def leverage_runner(monkeypatch): """ from contextlib import contextmanager from leverage.modules import tf, auth + from leverage.modules.tfrunner import TFRunner + + def skip_binary_validation(tf_runner): + """Accept the binary as is, without looking it up in PATH nor checking its version. + + Tests using this fixture mock the actual execution, and the binary is not necessarily + installed where the suite runs. `TFRunner` binary discovery is covered on its own in + tests/test_modules/test_tfrunner.py. + """ + tf_runner.binary_path = str(tf_runner.binary_input) @contextmanager def runner(leverage_directory): @@ -205,6 +216,9 @@ def runner(leverage_directory): monkeypatch.setattr(conf, "get_root_path", lambda: leverage_directory) monkeypatch.setattr(conf, "get_working_path", lambda: working_directory) + # Patch binary discovery so the tests do not depend on tofu being installed + monkeypatch.setattr(TFRunner, "_validate_binary", skip_binary_validation) + # Patch authentication functions to avoid SSO/credential checks monkeypatch.setattr(auth, "check_sso_token", lambda *args, **kwargs: None) monkeypatch.setattr(auth, "refresh_layer_credentials", lambda *args, **kwargs: None) diff --git a/tests/test_modules/test_kubectl.py b/tests/test_modules/test_kubectl.py index 9c0ea02..57c9935 100644 --- a/tests/test_modules/test_kubectl.py +++ b/tests/test_modules/test_kubectl.py @@ -45,7 +45,9 @@ def test_discover(leverage_project): } cli_runner = CliRunner() with cli_runner.isolated_filesystem(leverage_project) as leverage_project_folder: - with patch.object( + # The command only reaches _configure, so the binary does not need to be installed here. + # Runner binary discovery is covered on its own in tests/test_modules/test_runner.py. + with patch.object(kubectl_module.Runner, "_validate_binary", lambda runner: None), patch.object( kubectl_module, "_scan_clusters", return_value=[(leverage_project_folder, mocked_cluster_data)] ) as mkd_scan_clusters: with patch("simple_term_menu.TerminalMenu") as mkd_show: