Repository: getmoto/moto (python)

Issue:
Moto Secrets Manager rotation deviates from the real Secrets Manager
**Issue:**
When the rotation is iniated by moto, the version with stage AWSPENDING does not behave exactly the same as the real Secrets Manager secret.

I noticed that when the real Secrets Manager is called with rotate_secret, it will create a new version with stage AWSPENDING just like moto before it invokes the lambda for the `createSecret` step. However, this version does not have a secret value attached to it in the real Secrets Manager. 

If you call moto with `get_secret_value(VersionId=<the new versionid>, VersionStage="AWSPENDING")` moto will return a value (copied from AWSCURRENT). 

The real Secrets Manager will fail with `ResourceNotFoundException` regardless if you request either VersionId or VersionStage, or both. It seems like the version does not have a secret value attached to it at all.:

**Steps to reproduce:**
* Setup moto to create a Secret
* Setup a lambda in moto using the AWS lambda template (https://github.com/aws-samples/aws-secrets-manager-rotation-lambdas/blob/master/SecretsManagerRotationTemplate/lambda_function.py)
* Add env variable `SECRETS_MANAGER_ENDPOINT `or replace the code in the handler with the correct endpoint/remove it
* Replace the `NotImplementedError `statements in the lambda with pass or similar to make the function not throw
* Run the test with logging enabled

**Expected** result:
You should _not_ see `"createSecret: Successfully retrieved secret for..."` as this message indicates that the get_secret_value call succeeded on the first step as this is not how Secrets Manager behave.

Actual **result:**
You see the message (the version/stage AWSPENDING has a secret_value)


Pull request description:
SecretsManager: rotate_secret() should create AWSPending without a value
Fixes #8403 

When calling `rotate_secret()`, a `AWSPENDING` stage should exist, but it should not have a value. This can be verified by calling `get_secret_value()` during the rotation, at the `createSecret` step - it will throw an error.

The existing test that verifies the rotation works has been changed to reflect this:
 - During rotation, the current state is stored in a DynamoDB table
 - After the test finishes, we assert that we do indeed get a `ResourceNotFoundException` when calling `get_secret_value()`
 - (A lot more state is available, but we currently do not have any assertions to verify that they are correct)
 - The test has been modified to pass against AWS (having correct permissions, deleting everything afterwards, etc)

Patch:
diff --git a/moto/core/responses_custom_registry.py b/moto/core/responses_custom_registry.py
index 64c9422df..13db30a06 100644
--- a/moto/core/responses_custom_registry.py
+++ b/moto/core/responses_custom_registry.py
@@ -32,6 +32,23 @@ class CustomRegistry(responses.registries.FirstMatchRegistry):
         self._registered[response.method].append(response)
         return response
 
+    def replace(self, response: responses.BaseResponse) -> responses.BaseResponse:
+        registered = self._registered[response.method]
+        try:
+            index = registered.index(response)
+        except ValueError:
+            raise ValueError(f"Response is not registered for URL {response.url}")
+        registered[index] = response
+        return response
+
+    def remove(self, response: responses.BaseResponse) -> List[responses.BaseResponse]:
+        removed_responses = []
+        registered = self._registered[response.method]
+        while response in registered:
+            registered.remove(response)
+            removed_responses.append(response)
+        return removed_responses
+
     def reset(self) -> None:
         self._registered.clear()
 
diff --git a/moto/elasticache/exceptions.py b/moto/elasticache/exceptions.py
index 876406b4e..0266a151f 100644
--- a/moto/elasticache/exceptions.py
+++ b/moto/elasticache/exceptions.py
@@ -43,6 +43,26 @@ class PasswordRequired(ElastiCacheException):
         )
 
 
+class InvalidParameterValueException(ElastiCacheException):
+    code = 404
+
+    def __init__(self, message: str) -> None:
+        super().__init__(
+            "InvalidParameterValue",
+            message=message,
+        )
+
+
+class InvalidParameterCombinationException(ElastiCacheException):
+    code = 404
+
+    def __init__(self, message: str) -> None:
+        super().__init__(
+            "InvalidParameterCombination",
+            message=message,
+        )
+
+
 class UserAlreadyExists(ElastiCacheException):
     code = 404
 
diff --git a/moto/elasticache/models.py b/moto/elasticache/models.py
index ffade5875..6861d1bdd 100644
--- a/moto/elasticache/models.py
+++ b/moto/elasticache/models.py
@@ -12,10 +12,12 @@ from .exceptions import (
     CacheClusterAlreadyExists,
     CacheClusterNotFound,
     InvalidARNFault,
+    InvalidParameterCombinationException,
+    InvalidParameterValueException,
     UserAlreadyExists,
     UserNotFound,
 )
-from .utils import PAGINATION_MODEL
+from .utils import PAGINATION_MODEL, AuthenticationTypes
 
 
 class User(BaseModel):
@@ -29,10 +31,12 @@ class User(BaseModel):
         engine: str,
         no_password_required: bool,
         passwords: Optional[List[str]] = None,
+        authentication_type: Optional[str] = None,
     ):
         self.id = user_id
         self.name = user_name
         self.engine = engine
+
         self.passwords = passwords or []
         self.access_string = access_string
         self.no_password_required = no_password_required
@@ -41,6 +45,7 @@ class User(BaseModel):
         self.usergroupids: List[str] = []
         self.region = region
         self.arn = f"arn:{get_partition(self.region)}:elasticache:{self.region}:{account_id}:user:{self.id}"
+        self.authentication_type = authentication_type
 
 
 class CacheCluster(BaseModel):
@@ -161,9 +166,34 @@ class ElastiCacheBackend(BaseBackend):
         passwords: List[str],
         access_string: str,
         no_password_required: bool,
+        authentication_type: str,  # contain it to the str in the enums TODO
     ) -> User:
         if user_id in self.users:
             raise UserAlreadyExists
+
+        if authentication_type not in AuthenticationTypes._value2member_map_:
+            raise InvalidParameterValueException(
+                f"Input Authentication type: {authentication_type} is not in the allowed list: [password,no-password-required,iam]"
+            )
+
+        if (
+            no_password_required
+            and authentication_type != AuthenticationTypes.NOPASSWORD
+        ):
+            raise InvalidParameterCombinationException(
+                f"No password required flag is true but provided authentication type is {authentication_type}"
+            )
+
+        if passwords and authentication_type != AuthenticationTypes.PASSWORD:
+            raise InvalidParameterCombinationException(
+                f"Password field is not allowed with authentication type: {authentication_type}"
+            )
+
+        if not passwords and authentication_type == AuthenticationTypes.PASSWORD:
+            raise InvalidParameterCombinationException(
+                "A user with Authentication Mode: password, must have at least one password"
+            )
+
         user = User(
             account_id=self.account_id,
             region=self.region_name,
@@ -173,6 +203,7 @@ class ElastiCacheBackend(BaseBackend):
             passwords=passwords,
             access_string=access_string,
             no_password_required=no_password_required,
+            authentication_type=authentication_type,
         )
         self.users[user_id] = user
         return user
diff --git a/moto/elasticache/responses.py b/moto/elasticache/responses.py
index 225d2b7c8..82d6ecc0b 100644
--- a/moto/elasticache/responses.py
+++ b/moto/elasticache/responses.py
@@ -1,7 +1,12 @@
 from moto.core.responses import BaseResponse
 
-from .exceptions import PasswordRequired, PasswordTooShort
+from .exceptions import (
+    InvalidParameterCombinationException,
+    InvalidParameterValueException,
+    PasswordTooShort,
+)
 from .models import ElastiCacheBackend, elasticache_backends
+from .utils import AuthenticationTypes
 
 
 class ElastiCacheResponse(BaseResponse):
@@ -21,12 +26,41 @@ class ElastiCacheResponse(BaseResponse):
         user_name = params.get("UserName")
         engine = params.get("Engine")
         passwords = params.get("Passwords", [])
-        no_password_required = self._get_bool_param("NoPasswordRequired", False)
-        password_required = not no_password_required
-        if password_required and not passwords:
-            raise PasswordRequired
+        no_password_required = self._get_bool_param("NoPasswordRequired")
+        authentication_mode = params.get("AuthenticationMode")
+        authentication_type = "null"
+
+        if no_password_required is not None:
+            authentication_type = (
+                AuthenticationTypes.NOPASSWORD.value
+                if no_password_required
+                else AuthenticationTypes.PASSWORD.value
+            )
+
+        if passwords:
+            authentication_type = AuthenticationTypes.PASSWORD.value
+
+        if authentication_mode:
+            for key in authentication_mode.keys():
+                if key not in ["Type", "Passwords"]:
+                    raise InvalidParameterValueException(
+                        f'Unknown parameter in AuthenticationMode: "{key}", must be one of: Type, Passwords'
+                    )
+
+            authentication_type = authentication_mode.get("Type")
+            authentication_passwords = authentication_mode.get("Passwords", [])
+
+            if passwords and authentication_passwords:
+                raise InvalidParameterCombinationException(
+                    "Passwords provided via multiple arguments. Use only one argument"
+                )
+
+            # if passwords is empty, then we can use the authentication_passwords
+            passwords = passwords if passwords else authentication_passwords
+
         if any([len(p) < 16 for p in passwords]):
             raise PasswordTooShort
+
         access_string = params.get("AccessString")
         user = self.elasticache_backend.create_user(
             user_id=user_id,  # type: ignore[arg-type]
@@ -35,6 +69,7 @@ class ElastiCacheResponse(BaseResponse):
             passwords=passwords,
             access_string=access_string,  # type: ignore[arg-type]
             no_password_required=no_password_required,
+            authentication_type=authentication_type,
         )
         template = self.response_template(CREATE_USER_TEMPLATE)
         return template.render(user=user)
@@ -167,7 +202,9 @@ USER_TEMPLATE = """<UserId>{{ user.id }}</UserId>
       {% if user.no_password_required %}
       <Type>no-password</Type>
       {% else %}
-      <Type>password</Type>
+      <Type>{{ user.authentication_type }}</Type>
+      {% endif %}
+      {% if user.passwords %}
       <PasswordCount>{{ user.passwords|length }}</PasswordCount>
       {% endif %}
     </Authentication>
diff --git a/moto/elasticache/utils.py b/moto/elasticache/utils.py
index f0ac58c20..9377e997d 100644
--- a/moto/elasticache/utils.py
+++ b/moto/elasticache/utils.py
@@ -1,3 +1,5 @@
+from enum import Enum
+
 PAGINATION_MODEL = {
     "describe_cache_clusters": {
         "input_token": "marker",
@@ -6,3 +8,9 @@ PAGINATION_MODEL = {
         "unique_attribute": "cache_cluster_id",
     },
 }
+
+
+class AuthenticationTypes(str, Enum):
+    NOPASSWORD = "no-password-required"
+    PASSWORD = "password"
+    IAM = "iam"
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index a0c7acad0..4f83f7c8d 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -606,6 +606,7 @@ class SecretsManagerBackend(BaseBackend):
         version_stages: Optional[List[str]] = None,
         replica_regions: Optional[List[Dict[str, str]]] = None,
         force_overwrite: bool = False,
+        create_new_version: bool = False,
     ) -> Tuple[FakeSecret, bool]:
         if version_stages is None:
             version_stages = ["AWSCURRENT"]
@@ -633,7 +634,7 @@ class SecretsManagerBackend(BaseBackend):
 
             secret.update(description, tags, kms_key_id, last_changed_date=update_time)
 
-            if new_version:
+            if new_version or create_new_version:
                 if "AWSCURRENT" in version_stages:
                     secret.reset_default_version(secret_version, version_id)
                 else:
@@ -760,29 +761,21 @@ class SecretsManagerBackend(BaseBackend):
             pass
 
         if secret.versions:
-            old_secret_version = secret.versions[secret.default_version_id]  # type: ignore
-
             if client_request_token:
                 self._client_request_token_validator(client_request_token)
                 new_version_id = client_request_token
             else:
                 new_version_id = str(mock_random.uuid4())
 
-            # We add the new secret version as "pending". The previous version remains
-            # as "current" for now. Once we've passed the new secret through the lambda
-            # rotation function (if provided) we can then update the status to "current".
-            old_secret_version_secret_string = (
-                old_secret_version["secret_string"]
-                if "secret_string" in old_secret_version
-                else None
-            )
+            # We add a "pending" stage. The previous version remains as "current" for now.
+            # Caller is responsible for creating the new secret in the Lambda
             self._add_secret(
                 secret_id,
-                old_secret_version_secret_string,
                 description=secret.description,
                 tags=secret.tags,
                 version_id=new_version_id,
                 version_stages=["AWSPENDING"],
+                create_new_version=True,
             )
 
         secret.rotation_requested = True


Test patch:
diff --git a/tests/test_awslambda/__init__.py b/tests/test_awslambda/__init__.py
index 5dfd89f8a..015f8d34c 100644
--- a/tests/test_awslambda/__init__.py
+++ b/tests/test_awslambda/__init__.py
@@ -66,6 +66,13 @@ def lambda_aws_verified(func):
             kwargs["iam_role_arn"] = iam_role_arn
             resp = func(**kwargs)
         finally:
+            for policy in iam.list_attached_role_policies(RoleName=role_name)[
+                "AttachedPolicies"
+            ]:
+                iam.detach_role_policy(
+                    RoleName=role_name,
+                    PolicyArn=policy["PolicyArn"],
+                )
             iam.delete_role(RoleName=role_name)
 
         return resp
diff --git a/tests/test_core/test_request_mocking.py b/tests/test_core/test_request_mocking.py
index 3865ba679..d5ea67d88 100644
--- a/tests/test_core/test_request_mocking.py
+++ b/tests/test_core/test_request_mocking.py
@@ -1,8 +1,12 @@
+from unittest import SkipTest
+
 import boto3
 import pytest
 import requests
+from responses import Response
 
 from moto import mock_aws, settings
+from moto.core.models import responses_mock
 
 
 @mock_aws
@@ -48,3 +52,20 @@ def test_decorator_ordering() -> None:
 
     resp = requests.get(presigned_url)
     assert resp.status_code == 200
+
+
+@mock_aws()
+def test_replace_and_remove_mock() -> None:
+    if not settings.TEST_DECORATOR_MODE:
+        raise SkipTest("Only need to test responses mock in decorator mode")
+    rsp1 = Response(method="GET", url="http://example.com", body="test")
+    responses_mock.add(rsp1)
+
+    assert requests.get("http://example.com/").text == "test"
+
+    rsp2 = Response(method="GET", url="http://example.com", body="test2")
+    responses_mock.replace(rsp2)
+
+    assert requests.get("http://example.com/").text == "test2"
+
+    responses_mock.remove("GET", "https://example.com/")
diff --git a/tests/test_elasticache/test_elasticache.py b/tests/test_elasticache/test_elasticache.py
index d7b47934b..79c387d5a 100644
--- a/tests/test_elasticache/test_elasticache.py
+++ b/tests/test_elasticache/test_elasticache.py
@@ -1,6 +1,6 @@
 import boto3
 import pytest
-from botocore.exceptions import ClientError
+from botocore.exceptions import ClientError, ParamValidationError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
@@ -89,10 +89,279 @@ def test_create_user_without_password():
     assert err["Code"] == "InvalidParameterValue"
     assert (
         err["Message"]
-        == "No password was provided. If you want to create/update the user without password, please use the NoPasswordRequired flag."
+        == "Input Authentication type: null is not in the allowed list: [password,no-password-required,iam]"
     )
 
 
+@mock_aws
+def test_create_user_with_iam():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={"Type": "iam"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "iam"
+
+
+@mock_aws
+def test_create_user_invalid_authentication_type():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="User1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "invalidtype"},
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterValue"
+    assert (
+        err["Message"]
+        == "Input Authentication type: invalidtype is not in the allowed list: [password,no-password-required,iam]"
+    )
+
+
+@mock_aws
+def test_create_user_with_iam_with_passwords():
+    # IAM authentication mode should not come with password fields
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "iam"},
+            Passwords=["mysecretpassthatsverylong"],
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"] == "Password field is not allowed with authentication type: iam"
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_invalid_authmode_key():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"invalidkey": ""},
+        )
+    assert (
+        exc.value.kwargs["report"]
+        == 'Unknown parameter in AuthenticationMode: "invalidkey", must be one of: Type, Passwords'
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_empty_passwords_with_password():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            Passwords=[],
+            AuthenticationMode={"Type": "password"},
+        )
+    assert (
+        exc.value.kwargs["report"]
+        == "Invalid length for parameter Passwords, value: 0, valid min length: 1"
+    )
+
+
+# handled by botocore
+@mock_aws
+def test_create_user_with_empty_passwords_with_authmode_password():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ParamValidationError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "password", "Passwords": []},
+        )
+
+    assert (
+        exc.value.kwargs["report"]
+        == "Invalid length for parameter AuthenticationMode.Passwords, value: 0, valid min length: 1"
+    )
+
+
+@mock_aws
+def test_create_user_authmode_password_with_multiple_password_fields():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="on ~* +@all",
+            AuthenticationMode={"Type": "password", "Passwords": ["authmodepassword"]},
+            Passwords=["requestpassword"],
+            NoPasswordRequired=False,
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"]
+        == "Passwords provided via multiple arguments. Use only one argument"
+    )
+
+
+@mock_aws
+def test_create_user_with_authmode_password_without_passwords():
+    client = boto3.client("elasticache", region_name="us-east-1")
+    with pytest.raises(ClientError) as exc:
+        client.create_user(
+            UserId="user1",
+            UserName="user1",
+            Engine="Redis",
+            AccessString="?",
+            AuthenticationMode={"Type": "password"},
+        )
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterCombination"
+    assert (
+        err["Message"]
+        == "A user with Authentication Mode: password, must have at least one password"
+    )
+
+
+@mock_aws
+def test_create_user_with_authmode_no_password():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={"Type": "no-password-required"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "no-password-required"
+    assert (
+        "PasswordCount" not in resp["Authentication"]
+    )  # even though optional, we don't expect it for no-password-required
+
+
+@mock_aws
+def test_create_user_with_no_password_required_and_authmode_nopassword():
+    user_id = "user1"
+    client = boto3.client("elasticache", region_name="us-east-1")
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        NoPasswordRequired=True,
+        AuthenticationMode={"Type": "no-password-required"},
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "no-password"
+    assert (
+        "PasswordCount" not in resp["Authentication"]
+    )  # even though optional, we don't expect it for no-password-required
+
+
+@mock_aws
+def test_create_user_with_no_password_required_and_authmode_different():
+    for auth_mode in ["password", "iam"]:
+        client = boto3.client("elasticache", region_name="ap-southeast-1")
+        with pytest.raises(ClientError) as exc:
+            client.create_user(
+                UserId="user1",
+                UserName="user1",
+                Engine="Redis",
+                AccessString="on ~* +@all",
+                NoPasswordRequired=True,
+                AuthenticationMode={"Type": auth_mode},
+            )
+        err = exc.value.response["Error"]
+        assert err["Code"] == "InvalidParameterCombination"
+        assert (
+            err["Message"]
+            == f"No password required flag is true but provided authentication type is {auth_mode}"
+        )
+
+
+@mock_aws
+def test_create_user_with_authmode_password():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={
+            "Type": "password",
+            "Passwords": ["mysecretpassthatsverylong"],
+        },
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "password"
+    assert resp["Authentication"]["PasswordCount"] == 1
+
+
+@mock_aws
+def test_create_user_with_authmode_password_multiple():
+    # AuthenticationMode should be either 'password' or 'iam' or 'no-password'
+    client = boto3.client("elasticache", region_name="us-east-1")
+    user_id = "user1"
+    resp = client.create_user(
+        UserId=user_id,
+        UserName="User1",
+        Engine="Redis",
+        AccessString="on ~* +@all",
+        AuthenticationMode={
+            "Type": "password",
+            "Passwords": ["mysecretpassthatsverylong", "mysecretpassthatsverylong2"],
+        },
+    )
+
+    assert resp["Status"] == "active"
+    assert resp["Engine"] == "Redis"
+    assert resp["AccessString"] == "on ~* +@all"
+    assert resp["UserGroupIds"] == []
+    assert resp["Authentication"]["Type"] == "password"
+    assert resp["Authentication"]["PasswordCount"] == 2
+
+
 @mock_aws
 def test_create_user_twice():
     client = boto3.client("elasticache", region_name="ap-southeast-1")
diff --git a/tests/test_secretsmanager/__init__.py b/tests/test_secretsmanager/__init__.py
index cc6729aa9..fd749b0fb 100644
--- a/tests/test_secretsmanager/__init__.py
+++ b/tests/test_secretsmanager/__init__.py
@@ -17,27 +17,28 @@ def secretsmanager_aws_verified(func):
     """
 
     @wraps(func)
-    def pagination_wrapper():
+    def pagination_wrapper(**kwargs):
         allow_aws_request = (
             os.environ.get("MOTO_TEST_ALLOW_AWS_REQUEST", "false").lower() == "true"
         )
 
         if allow_aws_request:
-            return create_secret_and_execute(func)
+            return create_secret_and_execute(kwargs, func)
         else:
             with mock_aws():
-                return create_secret_and_execute(func)
+                return create_secret_and_execute(kwargs, func)
 
-    def create_secret_and_execute(func):
+    def create_secret_and_execute(kwargs, func):
         sm_client = boto3.client("secretsmanager", "us-east-1")
 
-        secret_arn = sm_client.create_secret(
+        secret = sm_client.create_secret(
             Name=f"moto_secret_{str(uuid4())[0:6]}",
             SecretString="old_secret",
-        )["ARN"]
+        )
         try:
-            return func(secret_arn)
+            kwargs["secret"] = secret
+            return func(**kwargs)
         finally:
-            sm_client.delete_secret(SecretId=secret_arn)
+            sm_client.delete_secret(SecretId=secret["ARN"])
 
     return pagination_wrapper
diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py
index fea06fa0a..6a853d964 100644
--- a/tests/test_secretsmanager/test_secretsmanager.py
+++ b/tests/test_secretsmanager/test_secretsmanager.py
@@ -15,6 +15,10 @@ from moto import mock_aws, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 from moto.secretsmanager.utils import SecretsManagerSecretIdentifier
 from moto.utilities.id_generator import TAG_KEY_CUSTOM_ID
+from tests import allow_aws_request
+from tests.test_awslambda import lambda_aws_verified
+from tests.test_awslambda.utilities import _process_lambda
+from tests.test_dynamodb import dynamodb_aws_verified
 
 from .. import DEFAULT_ACCOUNT_ID
 from . import secretsmanager_aws_verified
@@ -1110,20 +1114,47 @@ def test_rotate_secret_rotation_period_too_long():
 
 
 def get_rotation_zip_file():
-    from tests.test_awslambda.utilities import _process_lambda
+    endpoint = "" if allow_aws_request() else 'endpoint_url="http://motoserver:5000"'
 
-    func_str = """
+    func_str = (
+        """
 import boto3
 import json
+import os
 
 def lambda_handler(event, context):
     arn = event['SecretId']
     token = event['ClientRequestToken']
     step = event['Step']
 
-    client = boto3.client("secretsmanager", region_name="us-west-2", endpoint_url="http://motoserver:5000")
+    client = boto3.client("secretsmanager", region_name="us-east-1", """
+        + endpoint
+        + """)
     metadata = client.describe_secret(SecretId=arn)
-    value = client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
+    metadata.pop('LastChangedDate', None)
+    metadata.pop('LastAccessedDate', None)
+    metadata.pop('NextRotationDate', None)
+    metadata.pop('CreatedDate')
+    metadata.pop('ResponseMetadata')
+    print(metadata)
+    versions = client.list_secret_version_ids(SecretId=arn, IncludeDeprecated=True)["Versions"]
+    for v in versions:
+        v.pop('LastAccessedDate', None)
+        v.pop('CreatedDate', None)
+    print(versions)
+    try:
+        pending_value = client.get_secret_value(SecretId=arn, VersionId=token, VersionStage='AWSPENDING')
+        pending_value.pop('CreatedDate', None)
+        pending_value.pop('ResponseMetadata')
+    except Exception as e:
+        pending_value = str(e)
+    print(pending_value)
+    
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1", """
+        + endpoint
+        + """)
+    table = dynamodb.Table(os.environ["table_name"])
+    table.put_item(Item={"pk": step, "token": token, "metadata": metadata, "versions": versions, "pending_value": pending_value})
 
     if not metadata['RotationEnabled']:
         print("Secret %s is not enabled for rotation." % arn)
@@ -1140,24 +1171,19 @@ def lambda_handler(event, context):
         raise ValueError("Secret version %s not set as AWSPENDING for rotation of secret %s." % (token, arn))
 
     if step == 'createSecret':
-        try:
-            client.get_secret_value(SecretId=arn, VersionId=token, VersionStage='AWSPENDING')
-        except client.exceptions.ResourceNotFoundException:
-            client.put_secret_value(
-                SecretId=arn,
-                ClientRequestToken=token,
-                SecretString=json.dumps({'create': True}),
-                VersionStages=['AWSPENDING']
-            )
-
-    if step == 'setSecret':
         client.put_secret_value(
             SecretId=arn,
             ClientRequestToken=token,
             SecretString='UpdatedValue',
-            VersionStages=["AWSPENDING"]
+            VersionStages=['AWSPENDING']
         )
 
+    if step == 'setSecret':
+        # This method should set the AWSPENDING secret in the service that the secret belongs to.
+        # For example, if the secret is a database credential,
+        # this method should take the value of the AWSPENDING secret and set the user's password to this value in the database.
+        pass
+
     elif step == 'finishSecret':
         current_version = next(
             version
@@ -1177,60 +1203,103 @@ def lambda_handler(event, context):
             RemoveFromVersionId=token
         )
     """
+    )
     return _process_lambda(func_str)
 
 
-if settings.TEST_SERVER_MODE:
-
-    @mock_aws
-    def test_rotate_secret_using_lambda():
-        from tests.test_awslambda.utilities import get_role_name
-
-        # Passing a `RotationLambdaARN` value to `rotate_secret` should invoke lambda
-        lambda_conn = boto3.client(
-            "lambda", region_name="us-west-2", endpoint_url="http://localhost:5000"
+@pytest.mark.aws_verified
+@dynamodb_aws_verified()
+@lambda_aws_verified
+@secretsmanager_aws_verified
+def test_rotate_secret_using_lambda(secret=None, iam_role_arn=None, table_name=None):
+    role_name = iam_role_arn.split("/")[-1]
+    if not allow_aws_request() and not settings.TEST_SERVER_MODE:
+        raise SkipTest("Can only test this in ServerMode")
+
+    iam = boto3.client("iam", "us-east-1")
+    if allow_aws_request():
+        iam.attach_role_policy(
+            PolicyArn="arn:aws:iam::aws:policy/SecretsManagerReadWrite",
+            RoleName=role_name,
         )
-        func = lambda_conn.create_function(
-            FunctionName="testFunction",
-            Runtime="python3.11",
-            Role=get_role_name(),
-            Handler="lambda_function.lambda_handler",
-            Code={"ZipFile": get_rotation_zip_file()},
-            Description="Secret rotator",
-            Timeout=3,
-            MemorySize=128,
-            Publish=True,
+        # Testing this against AWS itself is a bit of pain
+        # Uncomment this to get more insights into what is happening during execution of the Lambda
+        # iam.attach_role_policy(
+        #    PolicyArn="arn:aws:iam::aws:policy/CloudWatchLogsFullAccess",
+        #    RoleName=role_name,
+        # )
+        iam.attach_role_policy(
+            PolicyArn="arn:aws:iam::aws:policy/AmazonDynamoDBFullAccess",
+            RoleName=role_name,
         )
 
-        secrets_conn = boto3.client(
-            "secretsmanager",
-            region_name="us-west-2",
-            endpoint_url="http://localhost:5000",
-        )
-        secret = secrets_conn.create_secret(
-            Name=DEFAULT_SECRET_NAME, SecretString="InitialValue"
-        )
-        initial_version = secret["VersionId"]
+    function_name = "moto_test_" + str(uuid4())[0:6]
 
-        rotated_secret = secrets_conn.rotate_secret(
-            SecretId=DEFAULT_SECRET_NAME,
-            RotationLambdaARN=func["FunctionArn"],
-            RotationRules={"AutomaticallyAfterDays": 30},
-        )
+    # Passing a `RotationLambdaARN` value to `rotate_secret` should invoke lambda
+    lambda_conn = boto3.client("lambda", region_name="us-east-1")
+    func = lambda_conn.create_function(
+        FunctionName=function_name,
+        Runtime="python3.11",
+        Role=iam_role_arn,
+        Handler="lambda_function.lambda_handler",
+        Code={"ZipFile": get_rotation_zip_file()},
+        Publish=True,
+        Environment={"Variables": {"table_name": table_name}},
+    )
+    lambda_conn.add_permission(
+        FunctionName=function_name,
+        StatementId="allow_secrets_manager",
+        Action="lambda:InvokeFunction",
+        Principal="secretsmanager.amazonaws.com",
+    )
+    lambda_conn.get_waiter("function_active_v2").wait(FunctionName=function_name)
+
+    secrets_conn = boto3.client("secretsmanager", region_name="us-east-1")
+
+    initial_version = secret["VersionId"]
+
+    rotated_secret = secrets_conn.rotate_secret(
+        SecretId=secret["ARN"],
+        RotationLambdaARN=func["FunctionArn"],
+        RotationRules={"AutomaticallyAfterDays": 30},
+        RotateImmediately=True,
+    )
 
-        # Ensure we received an updated VersionId from `rotate_secret`
-        assert rotated_secret["VersionId"] != initial_version
+    # Ensure we received an updated VersionId from `rotate_secret`
+    assert rotated_secret["VersionId"] != initial_version
 
+    secret_not_updated = True
+    while secret_not_updated:
         updated_secret = secrets_conn.get_secret_value(
-            SecretId=DEFAULT_SECRET_NAME, VersionStage="AWSCURRENT"
+            SecretId=secret["ARN"], VersionStage="AWSCURRENT"
         )
-        rotated_version = updated_secret["VersionId"]
+        if updated_secret["SecretString"] == "UpdatedValue":
+            secret_not_updated = False
+        else:
+            from time import sleep
+
+            sleep(5)
+    rotated_version = updated_secret["VersionId"]
+
+    assert initial_version != rotated_version
+
+    metadata = secrets_conn.describe_secret(SecretId=secret["ARN"])
+    assert metadata["VersionIdsToStages"][initial_version] == ["AWSPREVIOUS"]
+    assert metadata["VersionIdsToStages"][rotated_version] == ["AWSCURRENT"]
+    assert updated_secret["SecretString"] == "UpdatedValue"
+
+    lambda_conn.delete_function(FunctionName=function_name)
+
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+    items = dynamodb.Table(table_name).scan()["Items"]
+
+    create_secret = [i for i in items if i["pk"] == "createSecret"][0]
+    assert "(ResourceNotFoundException)" in create_secret["pending_value"]
+    assert create_secret["versions"][0]["VersionStages"] == ["AWSCURRENT"]
 
-        assert initial_version != rotated_version
-        metadata = secrets_conn.describe_secret(SecretId=DEFAULT_SECRET_NAME)
-        assert metadata["VersionIdsToStages"][initial_version] == ["AWSPREVIOUS"]
-        assert metadata["VersionIdsToStages"][rotated_version] == ["AWSCURRENT"]
-        assert updated_secret["SecretString"] == "UpdatedValue"
+    finish_secret = [i for i in items if i["pk"] == "finishSecret"][0]
+    assert finish_secret["pending_value"]["SecretString"] == "UpdatedValue"
+    assert finish_secret["pending_value"]["VersionStages"] == ["AWSPENDING"]
 
 
 @mock_aws
@@ -1887,49 +1956,49 @@ def test_update_secret_with_client_request_token():
 
 @secretsmanager_aws_verified
 @pytest.mark.aws_verified
-def test_update_secret_version_stage_manually(secret_arn=None):
+def test_update_secret_version_stage_manually(secret=None):
     sm_client = boto3.client("secretsmanager", "us-east-1")
     current_version = sm_client.put_secret_value(
-        SecretId=secret_arn,
+        SecretId=secret["ARN"],
         SecretString="previous_secret",
         VersionStages=["AWSCURRENT"],
     )["VersionId"]
 
     initial_secret = sm_client.get_secret_value(
-        SecretId=secret_arn, VersionStage="AWSCURRENT"
+        SecretId=secret["ARN"], VersionStage="AWSCURRENT"
     )
     assert initial_secret["VersionStages"] == ["AWSCURRENT"]
     assert initial_secret["SecretString"] == "previous_secret"
 
     token = str(uuid4())
     sm_client.put_secret_value(
-        SecretId=secret_arn,
+        SecretId=secret["ARN"],
         ClientRequestToken=token,
         SecretString="new_secret",
         VersionStages=["AWSPENDING"],
     )
 
     pending_secret = sm_client.get_secret_value(
-        SecretId=secret_arn, VersionStage="AWSPENDING"
+        SecretId=secret["ARN"], VersionStage="AWSPENDING"
     )
     assert pending_secret["VersionStages"] == ["AWSPENDING"]
     assert pending_secret["SecretString"] == "new_secret"
 
     sm_client.update_secret_version_stage(
-        SecretId=secret_arn,
+        SecretId=secret["ARN"],
         VersionStage="AWSCURRENT",
         MoveToVersionId=token,
         RemoveFromVersionId=current_version,
     )
 
     current_secret = sm_client.get_secret_value(
-        SecretId=secret_arn, VersionStage="AWSCURRENT"
+        SecretId=secret["ARN"], VersionStage="AWSCURRENT"
     )
     assert list(sorted(current_secret["VersionStages"])) == ["AWSCURRENT", "AWSPENDING"]
     assert current_secret["SecretString"] == "new_secret"
 
     previous_secret = sm_client.get_secret_value(
-        SecretId=secret_arn, VersionStage="AWSPREVIOUS"
+        SecretId=secret["ARN"], VersionStage="AWSPREVIOUS"
     )
     assert previous_secret["VersionStages"] == ["AWSPREVIOUS"]
     assert previous_secret["SecretString"] == "previous_secret"
@@ -1937,17 +2006,17 @@ def test_update_secret_version_stage_manually(secret_arn=None):
 
 @secretsmanager_aws_verified
 @pytest.mark.aws_verified
-def test_update_secret_version_stage_dont_specify_current_stage(secret_arn=None):
+def test_update_secret_version_stage_dont_specify_current_stage(secret=None):
     sm_client = boto3.client("secretsmanager", "us-east-1")
     current_version = sm_client.put_secret_value(
-        SecretId=secret_arn,
+        SecretId=secret["ARN"],
         SecretString="previous_secret",
         VersionStages=["AWSCURRENT"],
     )["VersionId"]
 
     token = str(uuid4())
     sm_client.put_secret_value(
-        SecretId=secret_arn,
+        SecretId=secret["ARN"],
         ClientRequestToken=token,
         SecretString="new_secret",
         VersionStages=["AWSPENDING"],
@@ -1956,7 +2025,7 @@ def test_update_secret_version_stage_dont_specify_current_stage(secret_arn=None)
     # Without specifying version that currently has stage AWSCURRENT
     with pytest.raises(ClientError) as exc:
         sm_client.update_secret_version_stage(
-            SecretId=secret_arn, VersionStage="AWSCURRENT", MoveToVersionId=token
+            SecretId=secret["ARN"], VersionStage="AWSCURRENT", MoveToVersionId=token
         )
     err = exc.value.response["Error"]
     assert err["Code"] == "InvalidParameterException"