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/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_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..57c9935 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,14 @@ 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)] + # 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: 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()