Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions aws_lambda_powertools/utilities/parameters/secrets.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@

return secret_value["SecretBinary"]

def _get_multiple(self, names: list[str], **sdk_options) -> dict[str, Any]: # type: ignore[override] # ty: ignore[invalid-method-override]

Check failure on line 133 in aws_lambda_powertools/utilities/parameters/secrets.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 18 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=aws-powertools_powertools-lambda-python&issues=AaFQI9aTRefphxp1sQDK&open=AaFQI9aTRefphxp1sQDK&pullRequest=8621
"""
Retrieve multiple secrets using AWS Secrets Manager batch_get_secret_value API

Expand Down Expand Up @@ -164,6 +164,7 @@
sdk_options.pop("SecretIdList", None)

secrets: dict[str, Any] = {}
requested_names = set(names)
next_token = None

# Handle pagination automatically
Expand All @@ -182,6 +183,8 @@
# Process successful secrets
for secret in response.get("SecretValues", []):
secret_name = secret["Name"]
if secret_name not in requested_names:
continue

# Extract secret value (SecretString or SecretBinary)
if "SecretString" in secret:
Expand Down
57 changes: 57 additions & 0 deletions tests/functional/parameters/_boto3/test_utilities_parameters.py
Original file line number Diff line number Diff line change
Expand Up @@ -3865,3 +3865,60 @@ def test_secrets_provider_get_multiple_with_json_transform_error(config):

with pytest.raises(parameters.TransformParameterError):
provider.get_multiple(secret_names, transform="binary", raise_on_transform_error=True)


@pytest.mark.parametrize("force_fetch", [False, True])
@pytest.mark.parametrize("secret_value", [{"SecretString": "value"}, {"SecretBinary": b"value"}])
def test_secrets_provider_get_multiple_excludes_prefix_matches(config, force_fetch, secret_value):
provider = parameters.SecretsProvider(boto_config=config)
names = ["db", "db-password"]
response = {
"SecretValues": [{"Name": name, **secret_value} for name in ["db", "db-password", "db2", "db-password-old"]],
}
expected_params = {"Filters": [{"Key": "name", "Values": names}]}
expected = dict.fromkeys(names, next(iter(secret_value.values())))

with stub.Stubber(provider.client) as stubber:
stubber.add_response("batch_get_secret_value", response, expected_params)
assert provider.get_multiple(names) == expected
if force_fetch:
stubber.add_response("batch_get_secret_value", response, expected_params)
assert provider.get_multiple(names, force_fetch=force_fetch) == expected
key = provider._build_cache_key(name="|".join(sorted(names)), transform=None, is_nested=True)
assert provider.fetch_from_cache(key) == expected
stubber.assert_no_pending_responses()


def test_secrets_provider_get_multiple_skips_prefix_only_page(config):
provider = parameters.SecretsProvider(boto_config=config)
names = ["db"]
expected_params = {"Filters": [{"Key": "name", "Values": names}]}
with stub.Stubber(provider.client) as stubber:
stubber.add_response(
"batch_get_secret_value",
{"SecretValues": [{"Name": "db-old", "SecretString": "not json"}], "NextToken": "next-page"},
expected_params,
)
stubber.add_response(
"batch_get_secret_value",
{"SecretValues": [{"Name": "db", "SecretString": '{"enabled": true}'}]},
{**expected_params, "NextToken": "next-page"},
)
assert provider.get_multiple(names, transform="json", raise_on_transform_error=True) == {
"db": {"enabled": True},
}
stubber.assert_no_pending_responses()


def test_secrets_provider_get_multiple_only_prefix_matches(config):
provider = parameters.SecretsProvider(boto_config=config)
with stub.Stubber(provider.client) as stubber:
stubber.add_response(
"batch_get_secret_value",
{"SecretValues": [{"Name": "db-old", "SecretString": "value"}]},
{"Filters": [{"Key": "name", "Values": ["db"]}]},
)
with pytest.raises(GetSecretError, match="No secrets found matching the provided names"):
provider.get_multiple(["db"])
assert not provider.store
stubber.assert_no_pending_responses()
Loading