diff --git a/localstack-core/localstack/testing/pytest/fixtures.py b/localstack-core/localstack/testing/pytest/fixtures.py index 95bc8b3db87bb..4da0016b8325f 100644 --- a/localstack-core/localstack/testing/pytest/fixtures.py +++ b/localstack-core/localstack/testing/pytest/fixtures.py @@ -1755,11 +1755,61 @@ def lambda_su_role(aws_client): run_safe(aws_client.iam.delete_policy(PolicyArn=policy_arn)) +@pytest.fixture +def create_iam_role_and_attach_policy(aws_client): + """ + Fixture that creates an IAM role with given role definition and predefined policy ARN. + + Use this fixture with AWS managed policies like 'AmazonS3ReadOnlyAccess' or 'AmazonKinesisFullAccess'. + """ + roles = [] + + def _inner(**kwargs: dict[str, any]) -> str: + """ + :param dict RoleDefinition: role definition document + :param str PolicyArn: policy ARN + :param str RoleName: role name (autogenerated if omitted) + :return: role ARN + """ + if "RoleName" not in kwargs: + kwargs["RoleName"] = f"test-role-{short_uid()}" + + role = kwargs["RoleName"] + role_policy = json.dumps(kwargs["RoleDefinition"]) + + result = aws_client.iam.create_role(RoleName=role, AssumeRolePolicyDocument=role_policy) + role_arn = result["Role"]["Arn"] + + policy_arn = kwargs["PolicyArn"] + aws_client.iam.attach_role_policy(PolicyArn=policy_arn, RoleName=role) + + roles.append(role) + return role_arn + + yield _inner + + for role in roles: + try: + aws_client.iam.delete_role(RoleName=role) + except Exception as exc: + LOG.debug("Error deleting IAM role '%s': %s", role, exc) + + @pytest.fixture def create_iam_role_with_policy(aws_client): + """ + Fixture that creates an IAM role with given role definition and policy definition. + """ roles = {} def _create_role_and_policy(**kwargs: dict[str, any]) -> str: + """ + :param dict RoleDefinition: role definition document + :param dict PolicyDefinition: policy definition document + :param str PolicyName: policy name (autogenerated if omitted) + :param str RoleName: role name (autogenerated if omitted) + :return: role ARN + """ if "RoleName" not in kwargs: kwargs["RoleName"] = f"test-role-{short_uid()}" role = kwargs["RoleName"] @@ -1781,8 +1831,14 @@ def _create_role_and_policy(**kwargs: dict[str, any]) -> str: yield _create_role_and_policy for role_name, policy_name in roles.items(): - aws_client.iam.delete_role_policy(RoleName=role_name, PolicyName=policy_name) - aws_client.iam.delete_role(RoleName=role_name) + try: + aws_client.iam.delete_role_policy(RoleName=role_name, PolicyName=policy_name) + except Exception as exc: + LOG.debug("Error deleting IAM role policy '%s' '%s': %s", role_name, policy_name, exc) + try: + aws_client.iam.delete_role(RoleName=role_name) + except Exception as exc: + LOG.debug("Error deleting IAM role '%s': %s", role_name, exc) @pytest.fixture