Repository: localstack/localstack
Pull request: [SecretsManager] Fixes
Addresses the following bug reports:
- It is possible to create secrets with invalid names which fail when you try to retrieve secrets for them with the message Secrets Manager can't find the specified secret value for staging label: AWSCURRENT
- Last accessed is not being updated when fetching secrets (getSecretValue)
- Last updated is not being updated when changing secrets (updateSecret)
- Description is being completely ignored, both when creating and updating secrets (it's always returned as an empty string)
It also fixes some date formats, mainly: CreatedDate, CreateDate, and LastChangedDate (last of which is related to bug above).

In addition, it fixes:
- many version stages update scenarios in leading to corruptions 
- secret rotation logic not cleaning up after rotation and not staging the version correctly before rotation
- prevents rotation fields to be updated if the provided lambda rotation arn is invalid
- the return type of update_secret_version_stage requests
- prevent return values of secretsmanager backend to include unset fields as done by aws in a number of scenarios


This adds a number of tests adhering to aws's behaviour, in scenarios this pr offers fixes; in short:
Modified:
- `test_create_and_update_secret`: added invalid secret id creation, added partial arn extraction for ids containing a dash at the end

Added:
- `test_rotate_secret_with_lambda_2`: tests rotating a secret twice using aws's base rotation logic and checks correct exceptions are thrown on creation
- `test_rotate_secret_lambda_invalid_arn`: tests rotation flags are not turned on if rotation starts with invalid lambda rotation arn
- `test_invalid_secret_name`: tests the creation/retrieval/update of secrets with invalid SecretIds
- `test_last_accessed_date`: tests the LastDateAccessed field is updated as expected in between retrievals of the secret
- `test_last_updated_date`: tests the LastDateUpdated field is updated as expected in between updates of the secret
- `test_update_secret_description`: tests the description is updated when instructed to do so
- `test_update_secret_version_stages_return_type`: tests the return type of update_secret_version_stages requests is correct (following the patch)
- `test_update_secret_version_stages_current_previous`: tests a scenario in which previously updating the secret with `AWSCURRENT` corrupted the secret's version stages
- `test_update_secret_version_stages_current_pending`: tests a scenario in which previously updating the secret with `AWSPENDING` corrupted the secret's version stages
- `test_update_secret_version_stages_current_pending_cycle`: tests a scenario in which previously updating the secret with `AWSPENDING` in a cycle from and to `AWSCURRENT` corrupted the secret's version stages
- `test_update_secret_version_stages_current_pending_cycle_custom_stages_1`: tests a scenario in which previously updating the secret with `AWSPENDING` with versions carrying custom version stages corrupted the secret's version stages
- `test_update_secret_version_stages_current_pending_cycle_custom_stages_2`: tests another cyclic scenario in which previously updating the secret with `AWSPENDING` and custom version stages corrupted the secret's version stages
- `test_non_versioning_version_stages_replacement`: tests the updating of the secret's version stages with custom version stages in a fashion leading to complete replacement of the versions stages of some previous version (previously not supported)
- `test_non_versioning_version_stages_no_replacement`: tests the updating of the secret's version stages with custom version stages in a fashion leading to a partial replacement of the versions stages of some previous version (previously not supported)
- `test_http_update_secret_with_missing_client_request_token`: tests secret update requests programmatically not specifying a client_request_token (this wasn't tested before)  


Problem statement:
bug: error on enable secrets manager rotation
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

**Trying enable secrets rotation**

```bash
awslocal secretsmanager rotate-secret \
    --secret-id arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq \
    --rotation-lambda-arn arn:aws:lambda:sa-east-1:000000000000:function:my-function \
    --rotation-rules AutomaticallyAfterDays=10
```
I get this error:
`An error occurred (ResourceNotFoundException) when calling the RotateSecret operation: Resource not found for ARN 'arn:aws:lambda:sa-east-1:000000000000:function:my-function'.`

**But the rotation looks enabled**

```bash
awslocal secretsmanager describe-secret --secret-id my-secret
```

```json
{
    "ARN": "arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq",
    "Name": "my-secret",
    "Description": "",
    "RotationEnabled": true,
    "RotationLambdaARN": "arn:aws:lambda:sa-east-1:000000000000:function:my-function",
    "RotationRules": {
        "AutomaticallyAfterDays": 10
    },
    "Tags": [],
    "VersionIdsToStages": {
        "e3815589-280c-49de-9c04-61ea475ce3a7": [
            "AWSCURRENT"
        ],
        "a9be99c5-6a54-401d-a3d1-d46915b38bdf": [
            "AWSPENDING"
        ]
    }
}
```

### Expected Behavior

No error

### How are you starting LocalStack?

With a docker-compose file

### Steps To Reproduce

#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)

```yaml
# DEBUG=1 docker-compose up

version: '3.1'

services:

  localstack:
    container_name: "${LOCALSTACK_DOCKER_NAME-localstack_main}"
    image: localstack/localstack
    network_mode: bridge
    restart: on-failure
    ports:
      # - "127.0.0.1:53:53"                # only required for Pro
      # - "127.0.0.1:53:53/udp"            # only required for Pro
      # - "127.0.0.1:443:443"              # only required for Pro
      # - "127.0.0.1:4510-4530:4510-4530"  # only required for Pro
      - "127.0.0.1:4566:4566"
      - "127.0.0.1:4571:4571"
    environment:
      - SERVICES=secretsmanager,ec2,sts,lambda,iam
      - DEBUG=${DEBUG-}
      # - DATA_DIR=${DATA_DIR-}
      - DATA_DIR=/private/localstack/data
      - LAMBDA_EXECUTOR=${LAMBDA_EXECUTOR-}
      - LOCALSTACK_API_KEY=${LOCALSTACK_API_KEY-}  # only required for Pro
      - HOST_TMP_FOLDER=${TMPDIR:-/tmp/}localstack
      - DOCKER_HOST=unix:///var/run/docker.sock
    volumes:
      - "${TMPDIR:-/tmp}/localstack:/tmp/localstack"
      - "/var/run/docker.sock:/var/run/docker.sock"
```

#### Client commands (e.g., AWS SDK code snippet, or sequence of "awslocal" commands)

```bash
awslocal secretsmanager create-secret \
--name my-secret \
--secret-string '{"key":"value"}'
```

```json
{
    "ARN": "arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq",
    "Name": "my-secret",
    "VersionId": "e3815589-280c-49de-9c04-61ea475ce3a7"
}
```

```bash
awslocal iam create-role \
--role-name role \
--assume-role-policy-document '{"Version":"2012-10-17","Statement":[{"Action":"sts:AssumeRole","Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"}}]}'
```

```json
{
    "Role": {
        "Path": "/",
        "RoleName": "role",
        "RoleId": "jboz4veok8wsqaj05n69",
        "Arn": "arn:aws:iam::000000000000:role/role",
        "CreateDate": "2022-01-12T19:19:49.606000+00:00",
        "AssumeRolePolicyDocument": {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": "sts:AssumeRole",
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "lambda.amazonaws.com"
                    }
                }
            ]
        },
        "MaxSessionDuration": 3600
    }
}
```

```bash
awslocal lambda create-function \
--function-name my-function \
--runtime nodejs12.x \
--role role \
--handler 'index.test' \
--zip-file fileb://lambda.zip
```

```json
{
    "FunctionName": "my-function",
    "FunctionArn": "arn:aws:lambda:sa-east-1:000000000000:function:my-function",
    "Runtime": "nodejs12.x",
    "Role": "role",
    "Handler": "index.test",
    "CodeSize": 337,
    "Description": "",
    "Timeout": 3,
    "LastModified": "2022-01-12T19:24:13.653+0000",
    "CodeSha256": "4Jp/nMvWr9wBA7U2EKjrvvlLyqnMNbf8HnDOXQY8Fkk=",
    "Version": "$LATEST",
    "VpcConfig": {},
    "TracingConfig": {
        "Mode": "PassThrough"
    },
    "RevisionId": "1db55c64-f593-42e6-aeac-1f8bde9722f9",
    "State": "Active",
    "LastUpdateStatus": "Successful",
    "PackageType": "Zip"
}
```

```bash
awslocal lambda add-permission \
--function-name "my-function" \
--statement-id "allow-secrets-manager" \
--action "lambda:InvokeFunction" \
--principal "secretsmanager.amazonaws.com" \
--source-arn "arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq"
``` 

```json
{
    "Statement": "{\"Sid\": \"allow-secrets-manager\", \"Effect\": \"Allow\", \"Action\": \"lambda:InvokeFunction\", \"Resource\": \"arn:aws:lambda:sa-east-1:000000000000:function:my-function\", \"Condition\": {\"ArnLike\": {\"AWS:SourceArn\": \"arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq\"}}, \"Principal\": {\"Service\": \"secretsmanager.amazonaws.com\"}}"
}
```

```bash
awslocal secretsmanager rotate-secret \
    --secret-id arn:aws:secretsmanager:sa-east-1:000000000000:secret:my-secret-mrUQmq \
    --rotation-lambda-arn arn:aws:lambda:sa-east-1:000000000000:function:my-function \
    --rotation-rules AutomaticallyAfterDays=10
```

```bash
An error occurred (ResourceNotFoundException) when calling the RotateSecret operation: Resource not found for ARN 'arn:aws:lambda:sa-east-1:000000000000:function:my-function'.
```

### Environment

```markdown
- OS:ubuntu 20.04
- LocalStack: docker tag latest
```


### Anything else?

_No response_

bug: SecretsManager ClientRequestToken Ignored on PutSecretValue
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Current Behavior

When creating a new version of a secret, the `ClientRequestToken` is not used as the `VersionId`.  It appears localstack generates a new `VersionId` every time the API is hit.

### Expected Behavior

> A unique identifier for the new version of the secret. If you use the Amazon Web Services CLI or one of the Amazon Web Services SDKs to call this operation, then you can leave this parameter empty because they generate a random UUID for you. If you don't use the SDK and instead generate a raw HTTP request to the Secrets Manager service endpoint, then you must generate a ClientRequestToken yourself for new versions and include that value in the request. This value helps ensure idempotency. Secrets Manager uses this value to prevent the accidental creation of duplicate versions if there are failures and retries during the Lambda rotation function processing. We recommend that you generate a UUID-type value to ensure uniqueness within the specified secret. If the ClientRequestToken value isn't already associated with a version of the secret then a new version of the secret is created. If a version with this value already exists and that version's SecretString or SecretBinary values are the same as those in the request then the request is ignored. The operation is idempotent. If a version with this value already exists and the version of the SecretString and SecretBinary values are different from those in the request, then the request fails because you can't modify a secret version. You can only create new versions to store new secret values.  _**This value becomes the VersionId of the new version.**_

Bold and italics added for emphasis.  I left the long description because I think the behavior when a duplicate `VersionId` is seen is also important.

### How are you starting LocalStack?

With a docker-compose file

### Steps To Reproduce

#### How are you starting localstack (e.g., `bin/localstack` command, arguments, or `docker-compose.yml`)

```yml
version: "3.1"

services:
  localstack:
    image: localstack/localstack:latest
    ports:
      - 4566:4566
```

#### Client commands (e.g., AWS SDK code snippet, or sequence of "awslocal" commands)

```typescript
await this.secretsManager.putSecretValue({
  SecretId: event.SecretId,
  ClientRequestToken: event.ClientRequestToken,
  SecretString: mySecret,
  VersionStages: [SecretStageLabel.Pending],
}).promise();
```

### Environment

```markdown
- OS: Ubuntu 20.04
- LocalStack: latest
```


### Anything else?

_No response_

Hints from issue discussion:
I'm also running into this bug and my configuration is pretty much the same, but I'm using the AWS SDK for Node.
There may be a similarly related bug where when creating a new secret value with the `AWSPENDING` stage label, it removes the `AWSCURRENT` stage label from the current version, which isn't supposed to happen until you issue the `UpdateSecretVersionStage` command.  I can create a separate bug report for that one if necessary, though.
What's the status of this, it seems like it got closed and reopened?

To add to this a little more,  I noticed another issue.  The `ClientRquestToken` appears to always be the same, whether I invoke the rotation in code via

```ts
await this.secretsManager.rotateSecret({
  SecretId: secretId,
  RotationLambdaARN: this.lambdaArn,
  RotationRules: {
    AutomaticallyAfterDays: rotationRate,
  },
}).promise();
```

or in the console with

```bash
awslocal secretsmanager rotate-secret --secret-id /secret/id
```

For this secret, I save the `ClientRequestToken` in a JSON blob.  When I run the above command, I see the CLI output a different VersionId each time, but when I look at the current secret value with 

```bash
awslocal secretsmanager get-secret-value --secret-id /secret/id --version-stage AWSCURRENT
```

The `ClientRequestToken` field I save in the JSON blob never changes.

I'm wondering if this is the root of the issue.  I've verified this functionality works as expected in AWS, so it shouldn't be something odd I'm doing.
Omitting a `ClientRequestToken` value in `PutSecretValue` request is not the direct source of the issues you've been experiencing. Some test replicating your description have been investigated as you can see here: https://github.com/localstack/localstack/pull/5545.

However, the second observation, about using `PutSecretValue` to update the `stage labels`, did in fact have some issues. These have been fixed, and you can find some test scenarios in the same PR https://github.com/localstack/localstack/pull/5545.

Could you please confirm this new issue now concerning rotateSecret, you are able to reproduce on the latest version of LocalStack?
I repulled the latest image yesterday right before I added the above comment.  It looks like now the `RemoveFromVersionId` field is being ignored.  Here's the result of DescribeSecret after UpdateSecretVersionStage:

```json
"VersionIdsToStages": {
   "a4651c21-055e-49b9-8eb1-f7a4fba5905c": [
       "AWSPREVIOUS"
   ],
   "4e34025e-8789-431a-8d92-93cf76d31401": [
       "AWSPENDING",
       "AWSCURRENT"
   ],
   "6bb0c339-4239-4728-ba36-027e229ab708": [
       "AWSPENDING",
       "AWSPREVIOUS"
   ]
 }
```

This was the command I ran

```bash
awslocal secretsmanager update-secret-version-stage --secret-id /cas/iam/jwt/c56fa25d-bfee-4a67-9a98-5900bf46a71a/key-pair --version-stage AWSCURRENT --move-to-version-id 4e34025e-8789-431a-8d92-93cf76d31401 --remove-from-version-id 6bb0c339-4239-4728-ba36-027e229ab708
```

> Omitting a ClientRequestToken value in PutSecretValue request is not the direct source of the issues you've been experiencing.

The issue as I understood it wasn't that I was omitting the `ClientRequestToken`.  According to the docs, if it's omitted, a unique one will be generated.  However, what I'm observing is that the same `ClientRequestToken` is being used every time the secret is generated.  For example, yesterday I rotated a secret 4 times (not including creating it).  Every time I rotated it, the field in the JSON blob I set based on the `ClientRequestToken` was always `c0c62618-b850-4a33-a553-25a60933842e`.  In production, when I rotate the secret, I see this value change.

I can open a separate ticket for that, though.
The API endpoint for updating the version stage also appears to return the wrong response now.  I'm now getting this error message

```
SyntaxError: Unexpected token / in JSON at position 0
    at JSON.parse (<anonymous>)
    at JsonParser.parse (./node_modules/aws-sdk/lib/json/parser.js:6:25)
    at Request.extractData (./node_modules/aws-sdk/lib/protocol/json.js:63:24)
    at Request.callListeners (./node_modules/aws-sdk/lib/sequential_executor.js:106:20)
    at Request.emit (./node_modules/aws-sdk/lib/sequential_executor.js:78:10)
    at Request.emit (./node_modules/aws-sdk/lib/request.js:686:14)
    at Request.transition (./node_modules/aws-sdk/lib/request.js:22:10)
    at AcceptorStateMachine.runTo (./node_modules/aws-sdk/lib/state_machine.js:14:12)
    at ./node_modules/aws-sdk/lib/state_machine.js:26:10
    at Request.<anonymous> (./node_modules/aws-sdk/lib/request.js:38:9)
    at Request.<anonymous> (./node_modules/aws-sdk/lib/request.js:688:12)
    at Request.callListeners (./node_modules/aws-sdk/lib/sequential_executor.js:116:18)
    at Request.emit (./node_modules/aws-sdk/lib/sequential_executor.js:78:10)
    at Request.emit (./node_modules/aws-sdk/lib/request.js:686:14)
    at Request.transition (./node_modules/aws-sdk/lib/request.js:22:10)
    at AcceptorStateMachine.runTo (./node_modules/aws-sdk/lib/state_machine.js:14:12)
    at ./node_modules/aws-sdk/lib/state_machine.js:26:10
    at Request.<anonymous> (./node_modules/aws-sdk/lib/request.js:38:9)
    at Request.<anonymous> (./node_modules/aws-sdk/lib/request.js:688:12)
    at Request.callListeners (./node_modules/aws-sdk/lib/sequential_executor.js:116:18)
    at callNextListener (./node_modules/aws-sdk/lib/sequential_executor.js:96:12)
    at IncomingMessage.onEnd (./node_modules/aws-sdk/lib/event_listeners.js:335:13)
    at IncomingMessage.emit (node:events:402:35)
    at IncomingMessage.emit (node:domain:475:12)
    at endReadableNT (node:internal/streams/readable:1343:12)
    at processTicksAndRejections (node:internal/process/task_queues:83:21)
```

I hit the API directly from Postman just to confirm it's returning the wrong type and can confirm.  It should be returning a JSON object, but it's returning a string (the secret ID) instead.

```
POST / HTTP/1.1
Host: localhost:4566
X-Amz-Target: secretsmanager.UpdateSecretVersionStage
Content-Type: application/x-amz-json-1.1
Accept-Encoding: identity
x-amz-date: 20220305T205748Z
X-Amz-Content-Sha256: beaead3198f7da1e70d03ab969765e0821b24fc913697e929e726aeaebf0eba3
Authorization: AWS4-HMAC-SHA256 Credential=foo/20220305/us-east-1/secretsmanager/aws4_request, SignedHeaders=host;x-amz-content-sha256;x-amz-date, Signature=2054116ac1ad937eb71599efb1cd65f31a37ac8fe8768b5f92a2f84010b56fd4
Content-Length: 244

{
    "SecretId": "/my/secret",
    "VersionStage": "AWSCURRENT",
    "RemoveFromVersionId": "ebd0e679-3ad9-445f-bcc2-27868aef4f2f",
    "MoveToVersionId": "e130a93f-0816-41c1-b330-9ab3cc988a2e"
}
```

The response:

```
/my/secret
```

Expected response (from the docs):

```
{
  "ARN":"arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestDatabaseSecret-a1b2c3",
  "Name":"MyTestDatabaseSecret"
}
```


Has there been any progress on this?  To summarize the errors as I see them right now

1. `UpdateSecretVersionStage` is returning an invalid response (returning a string instead of the expected JSON)
2. `ClientRequestToken` does not appear to be unique when generated by localstack.  This was tested using the Node AWS SDK and the AWS CLI.
3. `AWSPENDING` is not removed when the secret version stage is updated to `AWSCURRENT`.  This is something AWS does automatically and does not require a separate request.
4. localstack does not remove the `AWSPREVIOUS` version stage from the version that was previously marked as `AWSPREVIOUS` when changing the `AWSCURRENT` tag.  Again, this is something AWS automatically handles behind the scenes when updating the AWSCURRENT stage.

This is completely blocking me at this point because SecretsManager is completely unusable when using secret rotation in localstack.
Thank you for the patience and providing the summary above. These new points will soon get investigated.
I pulled the latest changes and at least point 1 above appears to be addressed, but point 2 is still an issue.  The issues with the version stages may also be addressed, but I haven't tested extensively.
@MEPalma It's been almost 2 months since this problem was reported and the original issue is still the issue I'm running into.  I tried to get around the issue of localstack reusing the same `ClientRequestToken` by specifying the `ClientRequestToken` when making the request to rotate the secret, but localstack seems to ignore the one provided (probably harkens back to the original report that " the `ClientRequestToken` is not used as the `VersionId`").

I've tried scanning through the code, but I can't find the actual implementation.  I'm not sure if it's open source, but if it is, could you point me to the right place?  I wouldn't mind contributing to the project to get unblocked.
With regards to the initial statement concerning the ClientRequestToken in PutSecretValue requests, this was fixed, and http tests following your description were added. Could you give me some feedback on those? They were crosschecked against AWS.

Concerning the latest descriptions, now pointing to version stages being corrupted, this is true, and is in the process of being fixed. It is unrelated to the ClientRequestToken missing, hence why you're probably still experiencing some issues. 
I would suggest waiting for the version stages update logic to be fixed, or attempt to better describe the workflow in which you do see the clientRequestToken being as issue.

Commit message: [minor]

Patch to review:
diff --git a/localstack/services/secretsmanager/provider.py b/localstack/services/secretsmanager/provider.py
index 0f52bfe6a5490..1d662e2601d96 100644
--- a/localstack/services/secretsmanager/provider.py
+++ b/localstack/services/secretsmanager/provider.py
@@ -1,11 +1,16 @@
+from __future__ import annotations
+
 import json
 import logging
-from typing import Dict, Optional
+import re
+import time
+import uuid
+from typing import Dict, Final, Optional, Union
 
 from moto.iam.policy_validation import IAMPolicyDocumentValidator
 from moto.secretsmanager import models as secretsmanager_models
-from moto.secretsmanager.exceptions import SecretNotFoundException
-from moto.secretsmanager.models import SecretsManagerBackend, secretsmanager_backends
+from moto.secretsmanager.exceptions import SecretNotFoundException, ValidationException
+from moto.secretsmanager.models import FakeSecret, SecretsManagerBackend, secretsmanager_backends
 from moto.secretsmanager.responses import SecretsManagerResponse
 
 from localstack.aws.api import RequestContext, ServiceResponse, handler
@@ -24,8 +29,11 @@
     GetResourcePolicyResponse,
     GetSecretValueRequest,
     GetSecretValueResponse,
+    InvalidParameterException,
+    InvalidRequestException,
     ListSecretVersionIdsRequest,
     ListSecretVersionIdsResponse,
+    NameType,
     PutResourcePolicyRequest,
     PutResourcePolicyResponse,
     PutSecretValueRequest,
@@ -34,10 +42,12 @@
     RemoveRegionsFromReplicationResponse,
     ReplicateSecretToRegionsRequest,
     ReplicateSecretToRegionsResponse,
+    ResourceNotFoundException,
     RestoreSecretRequest,
     RestoreSecretResponse,
     RotateSecretRequest,
     RotateSecretResponse,
+    SecretIdType,
     SecretsmanagerApi,
     StopReplicationToReplicaRequest,
     StopReplicationToReplicaResponse,
@@ -53,12 +63,18 @@
 from localstack.constants import TEST_AWS_ACCOUNT_ID
 from localstack.services.moto import call_moto, call_moto_with_request
 from localstack.utils.aws import aws_stack
-from localstack.utils.common import to_str
+from localstack.utils.patch import patch
 from localstack.utils.strings import short_uid
+from localstack.utils.time import today_no_time
+
+# Constants.
+AWSPREVIOUS: Final[str] = "AWSPREVIOUS"
+AWSPENDING: Final[str] = "AWSPENDING"
+AWSCURRENT: Final[str] = "AWSCURRENT"
 
 LOG = logging.getLogger(__name__)
 
-# maps key names to ARNs
+# Maps key names to ARNs.
 SECRET_ARN_STORAGE = {}
 
 
@@ -68,143 +84,458 @@ def __init__(self):
         apply_patches()
 
     @staticmethod
-    def _transform_context_secret_id(context: RequestContext) -> Optional[Dict]:
+    def _transform_context_secret_id(secret_id: SecretIdType) -> Optional[SecretIdType]:
         # If secret ARN ends with "-<randomId>" this is removed from the request for upstream compatibility.
-        data = json.loads(to_str(context.request.data or "{}"))
-        secret_id = data.get("SecretId", None)
-        if secret_id and ":" in secret_id:
+        t_secret_id = secret_id
+        if secret_id and re.match(r"^arn:", secret_id):
             arn = aws_stack.parse_arn(secret_id)
             aws_region = aws_stack.get_region()
             if arn["region"] != aws_region:
                 LOG.info(f'Expected request region "{aws_region}" for secret "{secret_id}"')
             resource_id = arn["resource"].split(":")[-1]
             if resource_id[-7] == "-":
-                data["SecretId"] = resource_id[:-7]
+                t_secret_id = resource_id[:-7]
             elif resource_id[-1] != "-":
-                data["SecretId"] += "-"
-            return data
-        return None
+                t_secret_id += "-"
+        return None if secret_id == t_secret_id else t_secret_id
+
+    @staticmethod
+    def _validate_secret_id(secret_id: SecretIdType) -> bool:
+        # The secret name can contain ASCII letters, numbers, and the following characters: /_+=.@-
+        return bool(re.match(r"^[A-Za-z0-9/_+=.@-]+\Z", secret_id))
+
+    @staticmethod
+    def _raise_if_invalid_secret_id(secret_id: Union[SecretIdType, NameType]):
+        # Patches moto's implementation for which secret_ids are not validated, by raising a ValidationException.
+        # Skips this check if the secret_id provided appears to be an arn (starting with 'arn:').
+        if not re.match(
+            r"^arn:", secret_id
+        ):  # Check if it appears to be an arn: so to skip secret_id check: delegate parsing of arn to handlers.
+            if not SecretsmanagerProvider._validate_secret_id(secret_id):
+                raise ValidationException(
+                    "Invalid name. Must be a valid name containing alphanumeric "
+                    "characters, or any of the following: -/_+=.@!"
+                )
 
     @staticmethod
-    def _call_moto_with_request_secret_id(context: RequestContext) -> ServiceResponse:
-        data_dict = SecretsmanagerProvider._transform_context_secret_id(context)
-        return call_moto_with_request(context, data_dict) if data_dict else call_moto(context)
+    def _call_moto_with_request_secret_id(
+        context: RequestContext, request: Dict
+    ) -> ServiceResponse:
+        t_secret_id = SecretsmanagerProvider._transform_context_secret_id(
+            request.get("SecretId", None)
+        )
+        if t_secret_id:
+            request.update({"SecretId": t_secret_id})
+            return call_moto_with_request(context, request)
+        return call_moto(context)
 
     @handler("CancelRotateSecret", expand=False)
     def cancel_rotate_secret(
         self, context: RequestContext, request: CancelRotateSecretRequest
     ) -> CancelRotateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("CreateSecret", expand=False)
     def create_secret(
         self, context: RequestContext, request: CreateSecretRequest
     ) -> CreateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["Name"])
+        return call_moto(context)
 
     @handler("DeleteResourcePolicy", expand=False)
     def delete_resource_policy(
         self, context: RequestContext, request: DeleteResourcePolicyRequest
     ) -> DeleteResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("DeleteSecret", expand=False)
     def delete_secret(
         self, context: RequestContext, request: DeleteSecretRequest
     ) -> DeleteSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("DescribeSecret", expand=False)
     def describe_secret(
         self, context: RequestContext, request: DescribeSecretRequest
     ) -> DescribeSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        res = self._call_moto_with_request_secret_id(context, request)
+        return res
 
     @handler("GetResourcePolicy", expand=False)
     def get_resource_policy(
         self, context: RequestContext, request: GetResourcePolicyRequest
     ) -> GetResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("GetSecretValue", expand=False)
     def get_secret_value(
         self, context: RequestContext, request: GetSecretValueRequest
     ) -> GetSecretValueResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ListSecretVersionIds", expand=False)
     def list_secret_version_ids(
         self, context: RequestContext, request: ListSecretVersionIdsRequest
     ) -> ListSecretVersionIdsResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("PutResourcePolicy", expand=False)
     def put_resource_policy(
         self, context: RequestContext, request: PutResourcePolicyRequest
     ) -> PutResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("PutSecretValue", expand=False)
     def put_secret_value(
         self, context: RequestContext, request: PutSecretValueRequest
     ) -> PutSecretValueResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RemoveRegionsFromReplication", expand=False)
     def remove_regions_from_replication(
         self, context: RequestContext, request: RemoveRegionsFromReplicationRequest
     ) -> RemoveRegionsFromReplicationResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ReplicateSecretToRegions", expand=False)
     def replicate_secret_to_regions(
         self, context: RequestContext, request: ReplicateSecretToRegionsRequest
     ) -> ReplicateSecretToRegionsResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RestoreSecret", expand=False)
     def restore_secret(
         self, context: RequestContext, request: RestoreSecretRequest
     ) -> RestoreSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RotateSecret", expand=False)
     def rotate_secret(
         self, context: RequestContext, request: RotateSecretRequest
     ) -> RotateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("StopReplicationToReplica", expand=False)
     def stop_replication_to_replica(
         self, context: RequestContext, request: StopReplicationToReplicaRequest
     ) -> StopReplicationToReplicaResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("TagResource", expand=False)
     def tag_resource(self, context: RequestContext, request: TagResourceRequest) -> None:
-        self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UntagResource", expand=False)
     def untag_resource(self, context: RequestContext, request: UntagResourceRequest) -> None:
-        self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UpdateSecret", expand=False)
     def update_secret(
         self, context: RequestContext, request: UpdateSecretRequest
     ) -> UpdateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UpdateSecretVersionStage", expand=False)
     def update_secret_version_stage(
         self, context: RequestContext, request: UpdateSecretVersionStageRequest
     ) -> UpdateSecretVersionStageResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ValidateResourcePolicy", expand=False)
     def validate_resource_policy(
         self, context: RequestContext, request: ValidateResourcePolicyRequest
     ) -> ValidateResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
+
+
+@patch(FakeSecret.__init__)
+def fake_secret__init__(fn, self, **kwargs):
+    fn(self, **kwargs)
+
+    # Fix time not including millis.
+    time_now = time.time()
+    if kwargs.get("last_changed_date", None):
+        self.last_changed_date = time_now
+    if kwargs.get("created_date", None):
+        self.created_date = time_now
+
+    # The last date that the secret value was retrieved.
+    # This value does not include the time.
+    # This field is omitted if the secret has never been retrieved.
+    self.last_accessed_date = None
+
+
+@patch(FakeSecret.update)
+def fake_secret_update(
+    fn, self, description=None, tags=None, kms_key_id=None, last_changed_date=None
+):
+    fn(self, description, tags, kms_key_id, last_changed_date)
+    if last_changed_date is not None:
+        self.last_changed_date = time.time()
+
+
+class FakeSecretVersionStore(dict):
+    def __setitem__(self, key, value):
+        self.put_version(key, value, time.time())
+
+    def put_version(self, version_id: str, version: Dict, create_date: Optional[float] = None):
+        if create_date and "createdate" in version:
+            version["createdate"] = create_date
+        super().__setitem__(version_id, version)
+
+
+@patch(FakeSecret.set_versions)
+def fake_secret_set_versions(_, self, versions):
+    self.versions = FakeSecretVersionStore()
+    for version_id, version in versions.items():
+        self.versions.put_version(version_id, version, self.created_date)
+
+
+@patch(SecretsManagerBackend.get_secret_value)
+def moto_smb_get_secret_value(fn, self, secret_id, version_id, version_stage):
+    res = fn(self, secret_id, version_id, version_stage)
+
+    secret_id = self.secrets[secret_id]
+    if secret_id:  # Redundant, we know from the response it exists: no exceptions.
+        secret_id.last_accessed_date = today_no_time()
+    else:
+        LOG.warning(
+            f'Expected Secret to exist on non failing GetSecretValue request for SecretId "{secret_id}"'
+        )
+
+    return res
+
+
+@patch(FakeSecret.to_dict)
+def fake_secret_to_dict(fn, self):
+    res_dict = fn(self)
+    if self.last_accessed_date:
+        res_dict["LastAccessedDate"] = self.last_accessed_date
+    return res_dict
+
+
+@patch(SecretsManagerBackend.update_secret)
+def backend_update_secret(
+    fn,
+    self,
+    secret_id,
+    description=None,
+    **kwargs,
+):
+    fn(self, secret_id, **kwargs)
+    secret = self.secrets[secret_id]
+
+    # Fix missing update of secret description.
+    # Secret exists if this point is reached.
+    if description is not None:
+        secret.description = description
+
+    return secret.to_short_dict()
+
+
+@patch(SecretsManagerResponse.update_secret)
+def response_update_secret(_, self):
+    secret_id = self._get_param("SecretId")
+    description = self._get_param("Description")
+    secret_string = self._get_param("SecretString")
+    secret_binary = self._get_param("SecretBinary")
+    client_request_token = self._get_param("ClientRequestToken")
+    kms_key_id = self._get_param("KmsKeyId")
+    return secretsmanager_backends[self.region].update_secret(
+        secret_id=secret_id,
+        description=description,
+        secret_string=secret_string,
+        secret_binary=secret_binary,
+        client_request_token=client_request_token,
+        kms_key_id=kms_key_id,
+    )
+
+
+@patch(SecretsManagerBackend.update_secret_version_stage)
+def backend_update_secret_version_stage(
+    fn, self, secret_id, version_stage, remove_from_version_id, move_to_version_id
+):
+    fn(self, secret_id, version_stage, remove_from_version_id, move_to_version_id)
+
+    secret = self.secrets[secret_id]
+    if version_stage == AWSCURRENT:
+        secret.default_version_id = move_to_version_id
+
+        # Ensure only one AWSPREVIOUS tagged version is in the pool.
+        # Remove secret versions with no version stages.
+        versions_no_stages = []
+        update_vid_set = {remove_from_version_id, move_to_version_id}
+        for version_id, version in secret.versions.items():
+            version_stages = version["version_stages"]
+            if version_id not in update_vid_set and AWSPREVIOUS in version_stages:
+                version_stages.remove(AWSPREVIOUS)
+                if not version_stages:
+                    versions_no_stages.append(version_id)
+
+        for version_no_stages in versions_no_stages:
+            del secret.versions[version_no_stages]
+
+    return json.dumps({"ARN": secret.arn, "Name": secret.name})
+
+
+@patch(FakeSecret.reset_default_version)
+def fake_secret_reset_default_version(fn, self, secret_version, version_id):
+    fn(self, secret_version, version_id)
+    # Remove versions with no version stages.
+    versions_no_stages = [
+        version_id for version_id, version in self.versions.items() if not version["version_stages"]
+    ]
+    for version_no_stages in versions_no_stages:
+        del self.versions[version_no_stages]
+
+
+@patch(FakeSecret.remove_version_stages_from_old_versions)
+def fake_secret_remove_version_stages_from_old_versions(fn, self, version_stages):
+    fn(self, version_stages)
+    # Remove versions with no version stages.
+    versions_no_stages = [
+        version_id for version_id, version in self.versions.items() if not version["version_stages"]
+    ]
+    for version_no_stages in versions_no_stages:
+        del self.versions[version_no_stages]
+
+
+@patch(SecretsManagerBackend.rotate_secret)
+def backend_rotate_secret(
+    _,
+    self,
+    secret_id,
+    client_request_token=None,
+    rotation_lambda_arn=None,
+    rotation_rules=None,
+):
+    rotation_days = "AutomaticallyAfterDays"
+
+    if not self._is_valid_identifier(secret_id):
+        raise SecretNotFoundException()
+
+    if self.secrets[secret_id].is_deleted():
+        raise InvalidRequestException(
+            "An error occurred (InvalidRequestException) when calling the RotateSecret operation: You tried to \
+            perform the operation on a secret that's currently marked deleted."
+        )
+
+    if rotation_lambda_arn:
+        if len(rotation_lambda_arn) > 2048:
+            msg = "RotationLambdaARN " "must <= 2048 characters long."
+            raise InvalidParameterException(msg)
+
+    if rotation_rules:
+        if rotation_days in rotation_rules:
+            rotation_period = rotation_rules[rotation_days]
+            if rotation_period < 1 or rotation_period > 1000:
+                msg = "RotationRules.AutomaticallyAfterDays " "must be within 1-1000."
+                raise InvalidParameterException(msg)
+
+    secret = self.secrets[secret_id]
+
+    # The rotation function must end with the versions of the secret in
+    # one of two states:
+    #
+    #  - The AWSPENDING and AWSCURRENT staging labels are attached to the
+    #    same version of the secret, or
+    #  - The AWSPENDING staging label is not attached to any version of the secret.
+    #
+    # If the AWSPENDING staging label is present but not attached to the same
+    # version as AWSCURRENT then any later invocation of RotateSecret assumes
+    # that a previous rotation request is still in progress and returns an error.
+    try:
+        version = next(
+            version
+            for version in secret.versions.values()
+            if AWSPENDING in version["version_stages"]
+        )
+        if AWSCURRENT in version["version_stages"]:
+            msg = "Previous rotation request is still in progress."
+            raise InvalidRequestException(msg)
+
+    except StopIteration:
+        # Pending is not present in any version
+        pass
+
+    if client_request_token:
+        self._client_request_token_validator(client_request_token)
+        new_version_id = client_request_token
+    else:
+        new_version_id = str(uuid.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".
+    self._add_secret(
+        secret_id,
+        None,
+        description=secret.description,
+        tags=secret.tags,
+        version_id=new_version_id,
+        version_stages=[AWSPENDING],
+    )
+    secret.rotation_lambda_arn = rotation_lambda_arn or ""
+    if rotation_rules:
+        secret.auto_rotate_after_days = rotation_rules.get(rotation_days, 0)
+    if secret.auto_rotate_after_days > 0:
+        secret.rotation_enabled = True
+
+    # Begin the rotation process for the given secret by invoking the lambda function.
+    if secret.rotation_lambda_arn:
+        from moto.awslambda.models import lambda_backends
+
+        lambda_backend = lambda_backends[self.region]
+
+        request_headers = {}
+        response_headers = {}
+
+        func = lambda_backend.get_function(secret.rotation_lambda_arn)
+        if not func:
+            msg = "Resource not found for ARN '{}'.".format(secret.rotation_lambda_arn)
+            raise ResourceNotFoundException(msg)
+
+        for step in ["create", "set", "test", "finish"]:
+            func.invoke(
+                json.dumps(
+                    {
+                        "Step": step + "Secret",
+                        "SecretId": secret.name,
+                        "ClientRequestToken": new_version_id,
+                    }
+                ),
+                request_headers,
+                response_headers,
+            )
+
+        secret.set_default_version_id(new_version_id)
+        version_stages = secret.versions[new_version_id]["version_stages"]
+        if AWSPENDING in version_stages:
+            version_stages.remove(AWSPENDING)
+    else:
+        secret.reset_default_version(secret.versions[new_version_id], new_version_id)
+        secret.versions[new_version_id]["version_stages"] = [AWSCURRENT]
+
+    return secret.to_short_dict()
 
 
 def secretsmanager_models_secret_arn(region, secret_id):
diff --git a/localstack/utils/time.py b/localstack/utils/time.py
index bf553dd0ee405..3592868d72cc0 100644
--- a/localstack/utils/time.py
+++ b/localstack/utils/time.py
@@ -1,5 +1,5 @@
 import time
-from datetime import datetime, timezone, tzinfo
+from datetime import date, datetime, timezone, tzinfo
 from typing import Optional
 
 TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S"
@@ -49,6 +49,10 @@ def now_utc(millis: bool = False) -> int:
     return now(millis, timezone.utc)
 
 
+def today_no_time() -> int:
+    return mktime(datetime.combine(date.today(), datetime.min.time()))
+
+
 def mktime(ts: datetime, millis: bool = False) -> int:
     if millis:
         return int(ts.timestamp() * 1000)
diff --git a/tests/integration/secretsmanager/__init__.py b/tests/integration/secretsmanager/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/secretsmanager/functions/__init__.py b/tests/integration/secretsmanager/functions/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/secretsmanager/functions/lambda_rotate_secret.py b/tests/integration/secretsmanager/functions/lambda_rotate_secret.py
new file mode 100644
index 0000000000000..98f74c4992ac5
--- /dev/null
+++ b/tests/integration/secretsmanager/functions/lambda_rotate_secret.py
@@ -0,0 +1,202 @@
+# Adepted from: https://github.com/aws-samples/aws-secrets-manager-rotation-lambdas/blob/252449551d58075f353444743e81a8c56d9f96db/SecretsManagerRotationTemplate/lambda_function.py
+# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0
+
+import logging
+import os
+
+import boto3
+
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+
+def secret_of_rotation_from_version_id(version_id: str) -> str:
+    return f"lambda_rotate_secret_rotation_{version_id}"
+
+
+def handler(event, context):
+    """Secrets Manager Rotation Template
+
+    This is a template for creating an AWS Secrets Manager rotation lambda
+
+    Args:
+        event (dict): Lambda dictionary of event parameters. These keys must include the following:
+            - SecretId: The secret ARN or identifier
+            - ClientRequestToken: The ClientRequestToken of the secret version
+            - Step: The rotation step (one of createSecret, setSecret, testSecret, or finishSecret)
+
+        context (LambdaContext): The Lambda runtime information
+
+    Raises:
+        ResourceNotFoundException: If the secret with the specified arn and stage does not exist
+
+        ValueError: If the secret is not properly configured for rotation
+
+        KeyError: If the event parameters do not contain the expected keys
+
+    """
+    # Setup the client
+    region_name = "us-east-1"
+    edge_port = 4566
+    protocol = "https" if os.environ.get("USE_SSL") else "http"
+    endpoing_url = f"{protocol}://{os.environ['LOCALSTACK_HOSTNAME']}:{edge_port}"
+    service_client = boto3.client(
+        "secretsmanager", region_name=region_name, endpoint_url=endpoing_url, verify=False
+    )
+
+    arn = event["SecretId"]
+    token = event["ClientRequestToken"]
+    step = event["Step"]
+
+    # Make sure the version is staged correctly
+    metadata = service_client.describe_secret(SecretId=arn)
+
+    if not metadata["RotationEnabled"]:
+        logger.error(f"Secret {arn} is not enabled for rotation")
+        raise ValueError(f"Secret {arn} is not enabled for rotation")
+    #
+    versions = metadata["VersionIdsToStages"]
+    if token not in versions:
+        logger.error(f"Secret version {token} has no stage for rotation of secret {arn}.")
+        raise ValueError(f"Secret version {token} has no stage for rotation of secret {arn}.")
+    if "AWSCURRENT" in versions[token]:
+        logger.info(f"Secret version {token} already set as AWSCURRENT for secret {arn}.")
+        return
+    elif "AWSPENDING" not in versions[token]:
+        logger.error(f"Secret version {token} not set as AWSPENDING for rotation of secret {arn}.")
+        raise ValueError(
+            f"Secret version {token} not set as AWSPENDING for rotation of secret {arn}."
+        )
+
+    if step == "createSecret":
+        create_secret(service_client, arn, token)
+
+    elif step == "setSecret":
+        set_secret(service_client, arn, token)
+
+    elif step == "testSecret":
+        test_secret(service_client, arn, token)
+
+    elif step == "finishSecret":
+        finish_secret(service_client, arn, token)
+
+    else:
+        raise ValueError("Invalid step parameter")
+
+
+def create_secret(service_client, arn, token):
+    """Create the secret
+
+    This method first checks for the existence of a secret for the passed in token. If one does not exist, it will generate a
+    new secret and put it with the passed in token.
+
+    Args:
+        service_client (client): The secrets manager service client
+
+        arn (string): The secret ARN or other identifier
+
+        token (string): The ClientRequestToken associated with the secret version
+
+    Raises:
+        ResourceNotFoundException: If the secret with the specified arn and stage does not exist
+
+    """
+    # Make sure the current secret exists
+    service_client.get_secret_value(SecretId=arn, VersionStage="AWSCURRENT")
+
+    # Now try to get the secret version, if that fails, put a new secret
+    try:
+        service_client.get_secret_value(SecretId=arn, VersionId=token, VersionStage="AWSPENDING")
+        logger.info(f"createSecret: Successfully retrieved secret for {arn}.")
+    except service_client.exceptions.ResourceNotFoundException:
+        # Generate a random password
+        passwd = secret_of_rotation_from_version_id(token)
+
+        # Put the secret
+        service_client.put_secret_value(
+            SecretId=arn,
+            ClientRequestToken=token,
+            SecretString=passwd,
+            VersionStages=["AWSPENDING"],
+        )
+        logger.info(
+            f"createSecret: Successfully put secret for ARN {arn} and version {token} with passwd {passwd}."
+        )
+
+
+def set_secret(service_client, arn, token):
+    """Set the secret
+
+    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.
+
+    Args:
+        service_client (client): The secrets manager service client
+
+        arn (string): The secret ARN or other identifier
+
+        token (string): The ClientRequestToken associated with the secret version
+
+    """
+    logger.info("lambda_rotate_secret: set_secret not implemented.")
+
+
+def test_secret(service_client, arn, token):
+    """Test the secret
+
+    This method should validate that the AWSPENDING secret works in the service that the secret belongs to. For example, if the secret
+    is a database credential, this method should validate that the user can login with the password in AWSPENDING and that the user has
+    all of the expected permissions against the database.
+
+    Args:
+        service_client (client): The secrets manager service client
+
+        arn (string): The secret ARN or other identifier
+
+        token (string): The ClientRequestToken associated with the secret version
+
+    """
+    logger.info("lambda_rotate_secret: test_secret not implemented.")
+
+
+def finish_secret(service_client, arn, token):
+    """Finish the secret
+
+    This method finalizes the rotation process by marking the secret version passed in as the AWSCURRENT secret.
+
+    Args:
+        service_client (client): The secrets manager service client
+
+        arn (string): The secret ARN or other identifier
+
+        token (string): The ClientRequestToken associated with the secret version
+
+    Raises:
+        ResourceNotFoundException: If the secret with the specified arn does not exist
+
+    """
+    # First describe the secret to get the current version
+    metadata = service_client.describe_secret(SecretId=arn)
+    current_version = None
+    for version in metadata["VersionIdsToStages"]:
+        if "AWSCURRENT" in metadata["VersionIdsToStages"][version]:
+            if version == token:
+                # The correct version is already marked as current, return
+                logger.info(
+                    f"finishSecret: Version {version} already marked as AWSCURRENT for {arn}"
+                )
+                return
+            current_version = version
+            break
+
+    # Finalize by staging the secret version current
+    service_client.update_secret_version_stage(
+        SecretId=arn,
+        VersionStage="AWSCURRENT",
+        MoveToVersionId=token,
+        RemoveFromVersionId=current_version,
+    )
+    logger.info(
+        f"finishSecret: Successfully set AWSCURRENT stage to version {token} for secret {arn}."
+    )
diff --git a/tests/integration/secretsmanager/test_secretsmanager.py b/tests/integration/secretsmanager/test_secretsmanager.py
new file mode 100644
index 0000000000000..682c7b1d38a73
--- /dev/null
+++ b/tests/integration/secretsmanager/test_secretsmanager.py
@@ -0,0 +1,1653 @@
+import json
+import os
+import uuid
+from datetime import datetime
+from typing import Dict, List, Optional
+
+import pytest
+import requests
+from py._code.code import ExceptionInfo
+
+from localstack.constants import TEST_AWS_ACCOUNT_ID
+from localstack.services.awslambda.lambda_utils import LAMBDA_RUNTIME_PYTHON36
+from localstack.utils import testutil
+from localstack.utils.aws import aws_stack
+from localstack.utils.strings import short_uid
+from localstack.utils.time import today_no_time
+from tests.integration.awslambda.test_lambda import TEST_LAMBDA_PYTHON_VERSION
+from tests.integration.secretsmanager.functions import lambda_rotate_secret
+
+RESOURCE_POLICY = {
+    "Version": "2012-10-17",
+    "Statement": [
+        {
+            "Effect": "Allow",
+            "Principal": {"AWS": "arn:aws:iam::%s:root" % TEST_AWS_ACCOUNT_ID},
+            "Action": "secretsmanager:GetSecretValue",
+            "Resource": "*",
+        }
+    ],
+}
+
+THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))
+TEST_LAMBDA_ROTATE_SECRET = os.path.join(THIS_FOLDER, "functions", "lambda_rotate_secret.py")
+
+
+class TestSecretsManager:
+    @pytest.fixture
+    def sm_client(self):
+        return aws_stack.create_external_boto_client("secretsmanager")
+
+    @pytest.mark.parametrize(
+        "secret_name, is_valid_partial_arn",
+        [
+            (f"s-{short_uid()}", True),
+            ("Valid/_+=.@-Name", True),
+            ("Valid/_+=.@-Name-a1b2", True),
+            ("Invalid/_+=.@-Name-a1b2c3", False),
+            ("Invalid/_+=.@-Name-a1b2c3-", False),
+        ],
+    )
+    def test_create_and_update_secret(
+        self, sm_client, secret_name: str, is_valid_partial_arn: bool
+    ):
+        description = "testing creation of secrets"
+        rs = sm_client.create_secret(
+            Name=secret_name,
+            SecretString="my_secret",
+            Description=description,
+        )
+        secret_arn = rs["ARN"]
+
+        assert len(secret_arn.rpartition("-")[-1]) == 6
+
+        rs = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs["Name"] == secret_name
+        assert rs["SecretString"] == "my_secret"
+        assert rs["ARN"] == secret_arn
+        assert isinstance(rs["CreatedDate"], datetime)
+
+        rs = sm_client.describe_secret(SecretId=secret_name)
+        assert rs["Name"] == secret_name
+        assert rs["ARN"] == secret_arn
+        assert isinstance(rs["CreatedDate"], datetime)
+        assert rs["Description"] == description
+
+        rs = sm_client.get_secret_value(SecretId=secret_arn)
+        assert rs["Name"] == secret_name
+        assert rs["SecretString"] == "my_secret"
+        assert rs["ARN"] == secret_arn
+
+        rs = sm_client.get_secret_value(SecretId=secret_arn[:-6])
+        assert rs["Name"] == secret_name
+        assert rs["SecretString"] == "my_secret"
+        assert rs["ARN"] == secret_arn
+
+        if is_valid_partial_arn:
+            rs = sm_client.get_secret_value(SecretId=secret_arn[:-7])
+            assert rs["Name"] == secret_name
+            assert rs["SecretString"] == "my_secret"
+            assert rs["ARN"] == secret_arn
+        else:
+            with pytest.raises(Exception) as resource_not_found:
+                sm_client.get_secret_value(SecretId=secret_arn[:-7])
+            assert resource_not_found.typename == "ResourceNotFoundException"
+
+        sm_client.put_secret_value(SecretId=secret_name, SecretString="new_secret")
+
+        rs = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs["Name"] == secret_name
+        assert rs["SecretString"] == "new_secret"
+
+        # update secret by ARN
+        rs = sm_client.update_secret(SecretId=secret_arn, KmsKeyId="test123", Description="d1")
+        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 200
+        assert rs["ARN"] == secret_arn
+
+        # clean up
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+
+    def test_call_lists_secrets_multiple_time(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        sm_client.create_secret(
+            Name=secret_name,
+            SecretString="my_secret",
+            Description="testing creation of secrets",
+        )
+
+        # call list_secrets multiple times
+        for i in range(3):
+            rs = sm_client.list_secrets()
+            secrets = [secret for secret in rs["SecretList"] if secret["Name"] == secret_name]
+            assert 1 == len(secrets)
+
+        # clean up
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+
+    def test_create_multi_secrets(self, sm_client):
+        secret_names = [short_uid(), short_uid(), short_uid()]
+        arns = []
+        for secret_name in secret_names:
+            rs = sm_client.create_secret(
+                Name=secret_name,
+                SecretString="my_secret_{}".format(secret_name),
+                Description="testing creation of secrets",
+            )
+            arns.append(rs["ARN"])
+
+        rs = sm_client.list_secrets()
+        secrets = {
+            secret["Name"]: secret["ARN"]
+            for secret in rs["SecretList"]
+            if secret["Name"] in secret_names
+        }
+
+        assert len(secrets.keys()) == len(secret_names)
+        for arn in arns:
+            assert arn in secrets.values()
+
+        # clean up
+        for secret_name in secret_names:
+            sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+
+    def test_get_random_exclude_characters_and_symbols(self, sm_client):
+        random_password = sm_client.get_random_password(
+            PasswordLength=120, ExcludeCharacters="xyzDje@?!."
+        )
+
+        assert len(random_password["RandomPassword"]) == 120
+        assert all(c not in "xyzDje@?!." for c in random_password["RandomPassword"])
+
+    def test_resource_policy(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+
+        sm_client.create_secret(
+            Name=secret_name,
+            SecretString="my_secret",
+            Description="testing creation of secrets",
+        )
+
+        sm_client.put_resource_policy(
+            SecretId=secret_name, ResourcePolicy=json.dumps(RESOURCE_POLICY)
+        )
+
+        rs = sm_client.get_resource_policy(SecretId=secret_name)
+
+        policy = json.loads(rs["ResourcePolicy"])
+
+        assert policy["Version"] == RESOURCE_POLICY["Version"]
+        assert policy["Statement"] == RESOURCE_POLICY["Statement"]
+
+        rs = sm_client.delete_resource_policy(SecretId=secret_name)
+        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+        # clean up
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+
+    def test_rotate_secret_with_lambda_1(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+
+        sm_client.create_secret(
+            Name=secret_name,
+            SecretString="my_secret",
+            Description="testing rotation of secrets",
+        )
+
+        function_name = "s-%s" % short_uid()
+        function_arn = testutil.create_lambda_function(
+            handler_file=TEST_LAMBDA_PYTHON_VERSION,
+            func_name=function_name,
+            runtime=LAMBDA_RUNTIME_PYTHON36,
+        )["CreateFunctionResponse"]["FunctionArn"]
+
+        response = sm_client.rotate_secret(
+            SecretId=secret_name,
+            RotationLambdaARN=function_arn,
+            RotationRules={
+                "AutomaticallyAfterDays": 1,
+            },
+            RotateImmediately=True,
+        )
+
+        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+        # clean up
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+        testutil.delete_lambda_function(function_name)
+
+    def test_rotate_secret_with_lambda_2(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+
+        cre_res = sm_client.create_secret(
+            Name=secret_name,
+            SecretString="init",
+            Description="testing rotation of secrets",
+        )
+        version_id_0 = cre_res["VersionId"]
+
+        function_name = f"rotate-func-{short_uid()}"
+        function_arn = testutil.create_lambda_function(
+            handler_file=TEST_LAMBDA_ROTATE_SECRET,
+            func_name=function_name,
+            runtime=LAMBDA_RUNTIME_PYTHON36,
+        )["CreateFunctionResponse"]["FunctionArn"]
+
+        rot_res = sm_client.rotate_secret(
+            SecretId=secret_name,
+            RotationLambdaARN=function_arn,
+            RotationRules={
+                "AutomaticallyAfterDays": 1,
+            },
+            RotateImmediately=True,
+        )
+        version_id_1 = rot_res["VersionId"]
+        assert version_id_0 != version_id_1
+
+        lst_res = sm_client.list_secret_version_ids(SecretId=secret_name)
+        versions = lst_res["Versions"]
+        assert len(versions) == 2
+
+        get_res_v0 = sm_client.get_secret_value(SecretId=secret_name, VersionId=version_id_0)
+        assert get_res_v0["VersionId"] == version_id_0
+        assert get_res_v0["SecretString"] == "init"
+
+        get_res_v1 = sm_client.get_secret_value(SecretId=secret_name, VersionId=version_id_1)
+        assert get_res_v1["VersionId"] == version_id_1
+        secret_string_1 = lambda_rotate_secret.secret_of_rotation_from_version_id(version_id_1)
+        assert get_res_v1["SecretString"] == secret_string_1
+
+        get_res = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_res["VersionId"] == version_id_1
+        assert get_res["SecretString"] == secret_string_1
+
+        rot_2_res = sm_client.rotate_secret(
+            SecretId=secret_name,
+            RotationLambdaARN=function_arn,
+            RotationRules={
+                "AutomaticallyAfterDays": 1,
+            },
+            RotateImmediately=True,
+        )
+        version_id_2 = rot_2_res["VersionId"]
+        assert len({version_id_0, version_id_1, version_id_2}) == 3
+
+        get_res = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_res["VersionId"] == version_id_2
+        secret_string_2 = lambda_rotate_secret.secret_of_rotation_from_version_id(version_id_2)
+        assert get_res["SecretString"] == secret_string_2
+
+        # clean up
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+        testutil.delete_lambda_function(function_name)
+
+    def test_put_secret_value_with_version_stages(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+
+        secret_string_v0: str = "secret_string_v0"
+        cr_v0_res = sm_client.create_secret(Name=secret_name, SecretString=secret_string_v0)
+        pv_v0_vid: str = cr_v0_res["VersionId"]
+
+        rs_get_curr = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs_get_curr["SecretString"] == secret_string_v0
+        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
+
+        secret_string_v1: str = "secret_string_v1"
+        version_stages_v1: ["str"] = ["SAMPLESTAGE1", "SAMPLESTAGE0"]
+        pv_v1_vid: str = str(uuid.uuid4())
+        pv_v1_res = sm_client.put_secret_value(
+            SecretId=secret_name,
+            SecretString=secret_string_v1,
+            VersionStages=version_stages_v1,
+            ClientRequestToken=pv_v1_vid,
+        )
+        assert pv_v1_res["VersionId"] == pv_v1_vid
+        assert pv_v1_res["VersionStages"] == version_stages_v1
+
+        rs_get_curr = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs_get_curr["VersionId"] == pv_v0_vid
+        assert rs_get_curr["SecretString"] == secret_string_v0
+        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
+
+        secret_string_v2: str = "secret_string_v2"
+        version_stages_v2: ["str"] = version_stages_v1
+        pv_v2_vid: str = str(uuid.uuid4())
+        pv_v2_res = sm_client.put_secret_value(
+            SecretId=secret_name,
+            SecretString=secret_string_v2,
+            VersionStages=version_stages_v2,
+            ClientRequestToken=pv_v2_vid,
+        )
+        assert pv_v2_res["VersionId"] == pv_v2_vid
+        assert pv_v2_res["VersionStages"] == version_stages_v2
+
+        rs_get_curr = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs_get_curr["VersionId"] == pv_v0_vid
+        assert rs_get_curr["SecretString"] == secret_string_v0
+        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
+
+        secret_string_v3: str = "secret_string_v3"
+        version_stages_v3: ["str"] = ["AWSPENDING"]
+        pv_v3_vid: str = str(uuid.uuid4())
+        pv_v3_res = sm_client.put_secret_value(
+            SecretId=secret_name,
+            SecretString=secret_string_v3,
+            VersionStages=version_stages_v3,
+            ClientRequestToken=pv_v3_vid,
+        )
+        assert pv_v3_res["VersionId"] == pv_v3_vid
+        assert pv_v3_res["VersionStages"] == version_stages_v3
+
+        rs_get_curr = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs_get_curr["VersionId"] == pv_v0_vid
+        assert rs_get_curr["SecretString"] == secret_string_v0
+        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
+
+        secret_string_v4: str = "secret_string_v4"
+        pv_v4_vid: str = str(uuid.uuid4())
+        pv_v4_res = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString=secret_string_v4, ClientRequestToken=pv_v4_vid
+        )
+        assert pv_v4_res["VersionId"] == pv_v4_vid
+        assert pv_v4_res["VersionStages"] == ["AWSCURRENT"]
+
+        rs_get_curr = sm_client.get_secret_value(SecretId=secret_name)
+        assert rs_get_curr["VersionId"] == pv_v4_vid
+        assert rs_get_curr["SecretString"] == secret_string_v4
+        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
+
+        sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+
+    @pytest.mark.parametrize(
+        "secret_name", ["Inv Name", " Inv Name", " Inv*Name? ", " Inv *?!]Name\\-"]
+    )
+    def test_invalid_secret_name(self, sm_client, secret_name: str):
+        def check_validation_exception(exc_info: ExceptionInfo):
+            error = exc_info.value.response["Error"]
+            error_code = error["Code"]
+            error_msg = error["Message"]
+            assert error_code == "ValidationException"
+            assert (
+                error_msg
+                == "Invalid name. Must be a valid name containing alphanumeric characters, or any of the following: -/_+=.@!"
+            )
+
+        # The secret name can contain ASCII letters, numbers, and the following characters: /_+=.@-
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.create_secret(Name=secret_name, SecretString="MySecretString")
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.describe_secret(SecretId=secret_name)
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.get_secret_value(SecretId=secret_name)
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.list_secret_version_ids(SecretId=secret_name, IncludeDeprecated=True)
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.put_secret_value(SecretId=secret_name, SecretString="MySecretString")
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.tag_resource(
+                SecretId=secret_name, Tags=[{"Key": "FirstTag", "Value": "SomeValue"}]
+            )
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.untag_resource(SecretId=secret_name, TagKeys=["FirstTag"])
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.update_secret(SecretId=secret_name, Description="MyNewDescription")
+        check_validation_exception(validation_exception)
+
+        with pytest.raises(Exception) as validation_exception:
+            sm_client.validate_resource_policy(
+                SecretId=secret_name,
+                ResourcePolicy='{\n"Version":"2012-10-17",\n"Statement":[{\n"Effect":"Allow",\n"Principal":{\n"AWS":"arn:aws:iam::123456789012:root"\n},\n"Action":"secretsmanager:GetSecretValue",\n"Resource":"*"\n}]\n}',
+            )
+        check_validation_exception(validation_exception)
+
+    def test_last_accessed_date(self, sm_client):
+        def last_accessed_scenario_1(fail_if_days_overlap: bool) -> bool:
+            secret_name = f"s-{short_uid()}"
+
+            sm_client.create_secret(Name=secret_name, SecretString="MySecretValue")
+
+            des = sm_client.describe_secret(SecretId=secret_name)
+            assert "LastAccessedDate" not in des
+
+            t0 = today_no_time()
+
+            sm_client.get_secret_value(SecretId=secret_name)
+            des = sm_client.describe_secret(SecretId=secret_name)
+            assert "LastAccessedDate" in des
+            lad_v0 = des["LastAccessedDate"]
+            assert isinstance(lad_v0, datetime)
+
+            sm_client.get_secret_value(SecretId=secret_name)
+            des = sm_client.describe_secret(SecretId=secret_name)
+            assert "LastAccessedDate" in des
+            lad_v1 = des["LastAccessedDate"]
+            assert isinstance(lad_v1, datetime)
+
+            if t0 == today_no_time() or fail_if_days_overlap:
+                assert lad_v0 == lad_v1
+                return True
+            else:
+                return False
+
+        if not last_accessed_scenario_1(
+            False
+        ):  # Test started yesterday and ended today (where relevant).
+            last_accessed_scenario_1(
+                True
+            )  # Replay today or allow failure (this should never take longer than a day).
+
+    def test_last_updated_date(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        sm_client.create_secret(Name=secret_name, SecretString="MySecretValue")
+
+        res = sm_client.describe_secret(SecretId=secret_name)
+        assert "LastChangedDate" in res
+        create_date = res["LastChangedDate"]
+        assert isinstance(create_date, datetime)
+
+        res = sm_client.get_secret_value(SecretId=secret_name)
+        assert create_date == res["CreatedDate"]
+
+        res = sm_client.describe_secret(SecretId=secret_name)
+        assert "LastChangedDate" in res
+        assert create_date == res["LastChangedDate"]
+
+        sm_client.update_secret(SecretId=secret_name, SecretString="MyNewSecretValue")
+
+        res = sm_client.describe_secret(SecretId=secret_name)
+        assert "LastChangedDate" in res
+        assert create_date < res["LastChangedDate"]
+        last_changed = res["LastChangedDate"]
+
+        sm_client.update_secret(SecretId=secret_name, SecretString="MyNewSecretValue")
+
+        res = sm_client.describe_secret(SecretId=secret_name)
+        assert "LastChangedDate" in res
+        assert last_changed < res["LastChangedDate"]
+
+        sm_client.update_secret(SecretId=secret_name, SecretString="MyVeryNewSecretValue")
+
+        res = sm_client.describe_secret(SecretId=secret_name)
+        assert "LastChangedDate" in res
+        assert create_date < res["LastChangedDate"]
+
+    def test_update_secret_description(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        secret_string_v0 = "MySecretString"
+        description_v0 = ""
+        sm_client.create_secret(Name=secret_name, SecretString=secret_string_v0)
+
+        des = sm_client.describe_secret(SecretId=secret_name)
+        assert des["Description"] == description_v0
+
+        description_v1 = "MyDescription"
+        sm_client.update_secret(SecretId=secret_name, Description=description_v1)
+
+        des = sm_client.describe_secret(SecretId=secret_name)
+        assert des["Description"] == description_v1
+
+        description_v2 = "MyNewDescription"
+        secret_string_v1 = "MyNewSecretString"
+        sm_client.update_secret(
+            SecretId=secret_name, SecretString=secret_string_v1, Description=description_v2
+        )
+
+        des = sm_client.describe_secret(SecretId=secret_name)
+        assert des["Description"] == description_v2
+
+        sm_client.update_secret(SecretId=secret_name, SecretString=secret_string_v1 * 2)
+        des = sm_client.describe_secret(SecretId=secret_name)
+        assert des["Description"] == description_v2
+
+    def test_update_secret_version_stages_return_type(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="Something1")
+        version_id_v0 = create["VersionId"]
+
+        put_pending_res = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="Something2", VersionStages=["AWSPENDING"]
+        )
+        version_id_v1 = put_pending_res["VersionId"]
+        assert version_id_v1 != version_id_v0
+
+        upd_res = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=version_id_v0,
+            MoveToVersionId=version_id_v1,
+            VersionStage="AWSCURRENT",
+        )
+        assert upd_res.keys() == {"Name", "ARN", "ResponseMetadata"}
+        assert upd_res["Name"] == create["Name"]
+        assert upd_res["ARN"] == create["ARN"]
+
+    def test_update_secret_version_stages_current_previous(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        secret_string_v0 = "secret_string_v0"
+        create = sm_client.create_secret(Name=secret_name, SecretString=secret_string_v0)
+        version_id_v0 = create["VersionId"]
+
+        lst_ids_v0 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        versions_v0 = lst_ids_v0["Versions"]
+        assert len(versions_v0) == 1
+        versions_v0_v0 = versions_v0[0]
+        assert versions_v0_v0["VersionId"] == version_id_v0
+        assert versions_v0_v0["VersionStages"] == ["AWSCURRENT"]
+
+        secret_string_v1 = "secret_string_v1"
+        update_v1 = sm_client.update_secret(SecretId=secret_name, SecretString=secret_string_v1)
+        version_id_v1 = update_v1["VersionId"]
+        assert version_id_v0 != version_id_v1
+
+        lst_ids_v1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        versions_v1 = lst_ids_v1["Versions"]
+        assert len(versions_v1) == 2
+        versions_v1_v0 = versions_v1[0]
+        assert versions_v1_v0["VersionId"] == version_id_v0
+        assert versions_v1_v0["VersionStages"] == ["AWSPREVIOUS"]
+        versions_v1_v1 = versions_v1[1]
+        assert versions_v1_v1["VersionId"] == version_id_v1
+        assert versions_v1_v1["VersionStages"] == ["AWSCURRENT"]
+
+    def test_update_secret_version_stages_current_pending(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="Something1")
+        version_id_v0 = create["VersionId"]
+
+        put_pending_res = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="Something2", VersionStages=["AWSPENDING"]
+        )
+        version_id_v1 = put_pending_res["VersionId"]
+        assert version_id_v1 != version_id_v0
+
+        list_ids_res = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(list_ids_res["Versions"]) == 2
+        list_ids_0 = list_ids_res["Versions"][0]
+        assert list_ids_0["VersionId"] == version_id_v0
+        assert list_ids_0["VersionStages"] == ["AWSCURRENT"]
+        list_ids_1 = list_ids_res["Versions"][1]
+        assert list_ids_1["VersionId"] == version_id_v1
+        assert list_ids_1["VersionStages"] == ["AWSPENDING"]
+
+        upd_res = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=version_id_v0,
+            MoveToVersionId=version_id_v1,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_res
+
+        list_ids_2_res = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(list_ids_2_res["Versions"]) == 2
+        list_ids_2_0 = list_ids_2_res["Versions"][0]
+        assert list_ids_2_0["VersionId"] == version_id_v0
+        assert list_ids_2_0["VersionStages"] == ["AWSPREVIOUS"]
+        list_ids_2_1 = list_ids_2_res["Versions"][1]
+        assert list_ids_2_1["VersionId"] == version_id_v1
+        assert list_ids_2_1["VersionStages"] == ["AWSPENDING", "AWSCURRENT"]
+
+        upd_2_res = sm_client.put_secret_value(SecretId=secret_name, SecretString="SS3")
+        version_id_v2 = upd_2_res["VersionId"]
+        assert len({version_id_v0, version_id_v1, version_id_v2}) == 3
+
+        list_ids_3_res = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(list_ids_3_res["Versions"]) == 2
+        list_ids_3_0 = list_ids_3_res["Versions"][0]
+        assert list_ids_3_0["VersionId"] == version_id_v1
+        assert list_ids_3_0["VersionStages"] == ["AWSPREVIOUS"]
+        list_ids_3_1 = list_ids_3_res["Versions"][1]
+        assert list_ids_3_1["VersionId"] == version_id_v2
+        assert list_ids_3_1["VersionStages"] == ["AWSCURRENT"]
+
+    def test_update_secret_version_stages_current_pending_cycle(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="S1")
+        vid_0 = create["VersionId"]
+
+        put_1 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S2", VersionStages=["AWSPENDING"]
+        )
+        vid_1 = put_1["VersionId"]
+        assert vid_1 != vid_0
+
+        lst_1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1["Versions"]) == 2
+        #
+        lst_1_v_0 = lst_1["Versions"][0]
+        assert lst_1_v_0["VersionId"] == vid_0
+        assert lst_1_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_1_v_1 = lst_1["Versions"][1]
+        assert lst_1_v_1["VersionId"] == vid_1
+        assert lst_1_v_1["VersionStages"] == ["AWSPENDING"]
+
+        get_1 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1["VersionId"] == vid_0
+        assert get_1["SecretString"] == "S1"
+        assert get_1["VersionStages"] == ["AWSCURRENT"]
+
+        upd_1 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_0,
+            MoveToVersionId=vid_1,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_1
+
+        lst_1_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1_u["Versions"]) == 2
+        #
+        lst_1_u_v_0 = lst_1_u["Versions"][0]
+        assert lst_1_u_v_0["VersionId"] == vid_0
+        assert lst_1_u_v_0["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_1_u_v_1 = lst_1_u["Versions"][1]
+        assert lst_1_u_v_1["VersionId"] == vid_1
+        assert lst_1_u_v_1["VersionStages"] == ["AWSPENDING", "AWSCURRENT"]
+
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_1
+        assert get_1_u["SecretString"] == "S2"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "AWSCURRENT"]
+
+        put_2 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S3", VersionStages=["AWSPENDING"]
+        )
+        vid_2 = put_2["VersionId"]
+        assert len({vid_0, vid_1, vid_2}) == 3
+
+        lst_2 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2["Versions"]) == 3
+        #
+        lst_2_v_0 = lst_2["Versions"][0]
+        assert lst_2_v_0["VersionId"] == vid_0
+        assert lst_2_v_0["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_2_v_1 = lst_2["Versions"][1]
+        assert lst_2_v_1["VersionId"] == vid_1
+        assert lst_2_v_1["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_2_v_2 = lst_2["Versions"][2]
+        assert lst_2_v_2["VersionId"] == vid_2
+        assert lst_2_v_2["VersionStages"] == ["AWSPENDING"]
+
+        get_2 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_2["VersionId"] == vid_1
+        assert get_2["SecretString"] == "S2"
+        assert get_2["VersionStages"] == ["AWSCURRENT"]
+
+        upd_2 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_1,
+            MoveToVersionId=vid_2,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_2
+
+        lst_2_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2_u["Versions"]) == 2
+        #
+        lst_2_u_v_0 = lst_2_u["Versions"][0]
+        assert lst_2_u_v_0["VersionId"] == vid_1
+        assert lst_2_u_v_0["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_2_u_v_1 = lst_2_u["Versions"][1]
+        assert lst_2_u_v_1["VersionId"] == vid_2
+        assert lst_2_u_v_1["VersionStages"] == ["AWSPENDING", "AWSCURRENT"]
+
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_2
+        assert get_1_u["SecretString"] == "S3"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "AWSCURRENT"]
+
+    def test_update_secret_version_stages_current_pending_cycle_custom_stages_1(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="S1")
+        vid_0 = create["VersionId"]
+
+        put_1 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S2", VersionStages=["AWSPENDING", "PUT1"]
+        )
+        vid_1 = put_1["VersionId"]
+        assert vid_1 != vid_0
+
+        lst_1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1["Versions"]) == 2
+        #
+        lst_1_v_0 = lst_1["Versions"][0]
+        assert lst_1_v_0["VersionId"] == vid_0
+        assert lst_1_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_1_v_1 = lst_1["Versions"][1]
+        assert lst_1_v_1["VersionId"] == vid_1
+        assert lst_1_v_1["VersionStages"] == ["AWSPENDING", "PUT1"]
+
+        get_1 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1["VersionId"] == vid_0
+        assert get_1["SecretString"] == "S1"
+        assert get_1["VersionStages"] == ["AWSCURRENT"]
+
+        upd_1 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_0,
+            MoveToVersionId=vid_1,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_1
+
+        lst_1_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1_u["Versions"]) == 2
+        #
+        lst_1_u_v_0 = lst_1_u["Versions"][0]
+        assert lst_1_u_v_0["VersionId"] == vid_0
+        assert lst_1_u_v_0["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_1_u_v_1 = lst_1_u["Versions"][1]
+        assert lst_1_u_v_1["VersionId"] == vid_1
+        assert lst_1_u_v_1["VersionStages"] == ["AWSPENDING", "PUT1", "AWSCURRENT"]
+
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_1
+        assert get_1_u["SecretString"] == "S2"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "PUT1", "AWSCURRENT"]
+
+        put_2 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S3", VersionStages=["AWSPENDING", "PUT2"]
+        )
+        vid_2 = put_2["VersionId"]
+        assert len({vid_0, vid_1, vid_2}) == 3
+
+        lst_2 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2["Versions"]) == 3
+        #
+        lst_2_v_0 = lst_2["Versions"][0]
+        assert lst_2_v_0["VersionId"] == vid_0
+        assert lst_2_v_0["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_2_v_1 = lst_2["Versions"][1]
+        assert lst_2_v_1["VersionId"] == vid_1
+        assert lst_2_v_1["VersionStages"] == ["PUT1", "AWSCURRENT"]
+        #
+        lst_2_v_2 = lst_2["Versions"][2]
+        assert lst_2_v_2["VersionId"] == vid_2
+        assert lst_2_v_2["VersionStages"] == ["AWSPENDING", "PUT2"]
+
+        get_2 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_2["VersionId"] == vid_1
+        assert get_2["SecretString"] == "S2"
+        assert get_2["VersionStages"] == ["PUT1", "AWSCURRENT"]
+
+        upd_2 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_1,
+            MoveToVersionId=vid_2,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_2
+
+        lst_2_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2_u["Versions"]) == 2
+        #
+        lst_2_u_v_0 = lst_2_u["Versions"][0]
+        assert lst_2_u_v_0["VersionId"] == vid_1
+        assert lst_2_u_v_0["VersionStages"] == ["PUT1", "AWSPREVIOUS"]
+        #
+        lst_2_u_v_1 = lst_2_u["Versions"][1]
+        assert lst_2_u_v_1["VersionId"] == vid_2
+        assert lst_2_u_v_1["VersionStages"] == ["AWSPENDING", "PUT2", "AWSCURRENT"]
+
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_2
+        assert get_1_u["SecretString"] == "S3"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "PUT2", "AWSCURRENT"]
+
+    def test_update_secret_version_stages_current_pending_cycle_custom_stages_2(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="SS")
+        vid_s = create["VersionId"]
+
+        put_0 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S1", VersionStages=["AWSCURRENT", "PUT0"]
+        )
+        vid_0 = put_0["VersionId"]
+        assert vid_0 != vid_s
+
+        lst_0 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_0["Versions"]) == 2
+        #
+        lst_0_v_s = lst_0["Versions"][0]
+        assert lst_0_v_s["VersionId"] == vid_s
+        assert lst_0_v_s["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_0_v_0 = lst_0["Versions"][1]
+        assert lst_0_v_0["VersionId"] == vid_0
+        assert lst_0_v_0["VersionStages"] == ["AWSCURRENT", "PUT0"]
+
+        put_1 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S2", VersionStages=["AWSPENDING", "PUT1"]
+        )
+        vid_1 = put_1["VersionId"]
+        assert len({vid_s, vid_0, vid_1}) == 3
+
+        lst_1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1["Versions"]) == 3
+        #
+        lst_1_v_s = lst_1["Versions"][0]
+        assert lst_1_v_s["VersionId"] == vid_s
+        assert lst_1_v_s["VersionStages"] == ["AWSPREVIOUS"]
+        #
+        lst_1_v_0 = lst_1["Versions"][1]
+        assert lst_1_v_0["VersionId"] == vid_0
+        assert lst_1_v_0["VersionStages"] == ["AWSCURRENT", "PUT0"]
+        #
+        lst_1_v_1 = lst_1["Versions"][2]
+        assert lst_1_v_1["VersionId"] == vid_1
+        assert lst_1_v_1["VersionStages"] == ["AWSPENDING", "PUT1"]
+
+        get_1 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1["VersionId"] == vid_0
+        assert get_1["SecretString"] == "S1"
+        assert get_1["VersionStages"] == ["AWSCURRENT", "PUT0"]
+
+        upd_1 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_0,
+            MoveToVersionId=vid_1,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_1
+
+        lst_1_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1_u["Versions"]) == 2
+        #
+        lst_1_u_v_0 = lst_1_u["Versions"][0]
+        assert lst_1_u_v_0["VersionId"] == vid_0
+        assert lst_1_u_v_0["VersionStages"] == ["PUT0", "AWSPREVIOUS"]
+        #
+        lst_1_u_v_1 = lst_1_u["Versions"][1]
+        assert lst_1_u_v_1["VersionId"] == vid_1
+        assert lst_1_u_v_1["VersionStages"] == ["AWSPENDING", "PUT1", "AWSCURRENT"]
+
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_1
+        assert get_1_u["SecretString"] == "S2"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "PUT1", "AWSCURRENT"]
+
+        put_2 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S3", VersionStages=["AWSPENDING", "PUT2"]
+        )
+        vid_2 = put_2["VersionId"]
+        assert len({vid_s, vid_0, vid_1, vid_2}) == 4
+
+        lst_2 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2["Versions"]) == 3
+        #
+        lst_2_v_0 = lst_2["Versions"][0]
+        assert lst_2_v_0["VersionId"] == vid_0
+        assert lst_2_v_0["VersionStages"] == ["PUT0", "AWSPREVIOUS"]
+        #
+        lst_2_v_1 = lst_2["Versions"][1]
+        assert lst_2_v_1["VersionId"] == vid_1
+        assert lst_2_v_1["VersionStages"] == ["PUT1", "AWSCURRENT"]
+        #
+        lst_2_v_2 = lst_2["Versions"][2]
+        assert lst_2_v_2["VersionId"] == vid_2
+        assert lst_2_v_2["VersionStages"] == ["AWSPENDING", "PUT2"]
+
+        get_2 = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_2["VersionId"] == vid_1
+        assert get_2["SecretString"] == "S2"
+        assert get_2["VersionStages"] == ["PUT1", "AWSCURRENT"]
+
+        upd_2 = sm_client.update_secret_version_stage(
+            SecretId=secret_name,
+            RemoveFromVersionId=vid_1,
+            MoveToVersionId=vid_2,
+            VersionStage="AWSCURRENT",
+        )
+        assert "VersionId" not in upd_2
+
+        lst_2_u = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_2_u["Versions"]) == 3
+        #
+        lst_2_u_v_0 = lst_2_u["Versions"][0]
+        assert lst_2_u_v_0["VersionId"] == vid_0
+        assert lst_2_u_v_0["VersionStages"] == ["PUT0"]
+        #
+        lst_2_u_v_1 = lst_2_u["Versions"][1]
+        assert lst_2_u_v_1["VersionId"] == vid_1
+        assert lst_2_u_v_1["VersionStages"] == ["PUT1", "AWSPREVIOUS"]
+        #
+        lst_2_u_v_2 = lst_2_u["Versions"][2]
+        assert lst_2_u_v_2["VersionId"] == vid_2
+        assert lst_2_u_v_2["VersionStages"] == ["AWSPENDING", "PUT2", "AWSCURRENT"]
+        #
+        get_1_u = sm_client.get_secret_value(SecretId=secret_name)
+        assert get_1_u["VersionId"] == vid_2
+        assert get_1_u["SecretString"] == "S3"
+        assert get_1_u["VersionStages"] == ["AWSPENDING", "PUT2", "AWSCURRENT"]
+
+    def test_non_versioning_version_stages_replacement(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="S0")
+        vid_s = create["VersionId"]
+
+        put_0 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S1", VersionStages=["one", "two", "three"]
+        )
+        vid_0 = put_0["VersionId"]
+        assert vid_0 != vid_s
+
+        lst_0 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_0["Versions"]) == 2
+        #
+        lst_0_v_0 = lst_0["Versions"][0]
+        assert lst_0_v_0["VersionId"] == vid_s
+        assert lst_0_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_0_v_1 = lst_0["Versions"][1]
+        assert lst_0_v_1["VersionId"] == vid_0
+        assert lst_0_v_1["VersionStages"] == ["one", "two", "three"]
+
+        put_1 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S2", VersionStages=["one", "two", "three", "four"]
+        )
+        vid_1 = put_1["VersionId"]
+        assert len({vid_s, vid_0, vid_1}) == 3
+
+        lst_1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1["Versions"]) == 2
+        #
+        lst_1_v_0 = lst_1["Versions"][0]
+        assert lst_1_v_0["VersionId"] == vid_s
+        assert lst_1_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_1_v_1 = lst_1["Versions"][1]
+        assert lst_1_v_1["VersionId"] == vid_1
+        assert lst_1_v_1["VersionStages"] == ["one", "two", "three", "four"]
+
+    def test_non_versioning_version_stages_no_replacement(self, sm_client):
+        secret_name = f"s-{short_uid()}"
+        create = sm_client.create_secret(Name=secret_name, SecretString="S0")
+        vid_s = create["VersionId"]
+
+        put_0 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S1", VersionStages=["one", "two", "three"]
+        )
+        vid_0 = put_0["VersionId"]
+        assert vid_0 != vid_s
+
+        lst_0 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_0["Versions"]) == 2
+        #
+        lst_0_v_0 = lst_0["Versions"][0]
+        assert lst_0_v_0["VersionId"] == vid_s
+        assert lst_0_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_0_v_1 = lst_0["Versions"][1]
+        assert lst_0_v_1["VersionId"] == vid_0
+        assert lst_0_v_1["VersionStages"] == ["one", "two", "three"]
+
+        put_1 = sm_client.put_secret_value(
+            SecretId=secret_name, SecretString="S2", VersionStages=["one", "two", "four"]
+        )
+        vid_1 = put_1["VersionId"]
+        assert len({vid_s, vid_0, vid_1}) == 3
+
+        lst_1 = sm_client.list_secret_version_ids(SecretId=secret_name)
+        assert len(lst_1["Versions"]) == 3
+        #
+        lst_1_v_0 = lst_1["Versions"][0]
+        assert lst_1_v_0["VersionId"] == vid_s
+        assert lst_1_v_0["VersionStages"] == ["AWSCURRENT"]
+        #
+        lst_1_v_1 = lst_1["Versions"][1]
+        assert lst_1_v_1["VersionId"] == vid_0
+        assert lst_1_v_1["VersionStages"] == ["three"]
+        #
+        lst_1_v_2 = lst_1["Versions"][2]
+        assert lst_1_v_2["VersionId"] == vid_1
+        assert lst_1_v_2["VersionStages"] == ["one", "two", "four"]
+
+    @staticmethod
+    def secretsmanager_http_json_headers(amz_target: str) -> Dict:
+        headers = aws_stack.mock_aws_request_headers("secretsmanager")
+        headers["X-Amz-Target"] = amz_target
+        return headers
+
+    def secretsmanager_http_json_post(self, amz_target: str, http_body: json) -> requests.Response:
+        ep_url: str = aws_stack.get_local_service_url("secretsmanager")
+        http_headers: Dict = self.secretsmanager_http_json_headers(amz_target)
+        return requests.post(ep_url, headers=http_headers, data=json.dumps(http_body))
+
+    def secretsmanager_http_create_secret_string(
+        self, secret_name: str, secret_string: str
+    ) -> requests.Response:
+        http_body: json = {"Name": secret_name, "SecretString": secret_string}
+        return self.secretsmanager_http_json_post("secretsmanager.CreateSecret", http_body)
+
+    @staticmethod
+    def secretsmanager_http_create_secret_string_val_res(
+        res: requests.Response, secret_name: str
+    ) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        return res_json
+
+    def secretsmanager_http_delete_secret(self, secret_id: str) -> requests.Response:
+        http_body: json = {"SecretId": secret_id}
+        return self.secretsmanager_http_json_post("secretsmanager.DeleteSecret", http_body)
+
+    @staticmethod
+    def secretsmanager_http_delete_secret_val_res(res: requests.Response, secret_id: str) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_id
+        return res_json
+
+    def secretsmanager_http_get_secret_value(self, secret_id: str) -> requests.Response:
+        http_body: json = {"SecretId": secret_id}
+        return self.secretsmanager_http_json_post("secretsmanager.GetSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_get_secret_value_val_res(
+        res: requests.Response, secret_name: str, secret_string: str, version_id: str
+    ) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        assert res_json["SecretString"] == secret_string
+        assert res_json["VersionId"] == version_id
+        return res_json
+
+    def secretsmanager_http_get_secret_value_with(
+        self, secret_id: str, version_stage: str
+    ) -> requests.Response:
+        http_body: json = {"SecretId": secret_id, "VersionStage": version_stage}
+        return self.secretsmanager_http_json_post("secretsmanager.GetSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_get_secret_value_with_val_res(
+        res: requests.Response,
+        secret_name: str,
+        secret_string: str,
+        version_id: str,
+        version_stage: str,
+    ) -> json:
+        res_json = TestSecretsManager.secretsmanager_http_get_secret_value_val_res(
+            res, secret_name, secret_string, version_id
+        )
+        assert res_json["VersionStages"] == [version_stage]
+        return res_json
+
+    def secretsmanager_http_list_secret_version_ids(self, secret_id: str) -> requests.Response:
+        http_body: json = {"SecretId": secret_id}
+        return self.secretsmanager_http_json_post("secretsmanager.ListSecretVersionIds", http_body)
+
+    @staticmethod
+    def secretsmanager_http_list_secret_version_ids_val_res(
+        res: requests.Response, secret_name: str, versions: json
+    ) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        res_versions: [json] = res_json["Versions"]
+        assert len(res_versions) == len(versions)
+        assert len(set([rv["VersionId"] for rv in res_versions])) == len(res_versions)
+        assert len(set([v["VersionId"] for v in versions])) == len(versions)
+        for version in versions:
+            vs_in_res: [json] = list(
+                filter(lambda rv: rv["VersionId"] == version["VersionId"], res_versions)
+            )
+            assert len(vs_in_res) == 1
+            v_in_res = vs_in_res[0]
+            assert v_in_res["VersionStages"] == version["VersionStages"]
+        return res_json
+
+    def secretsmanager_http_put_secret_value(
+        self, secret_id: str, secret_string: str
+    ) -> requests.Response:
+        http_body: json = {
+            "SecretId": secret_id,
+            "SecretString": secret_string,
+        }
+        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_put_secret_value_val_res(
+        res: requests.Response, secret_name: str
+    ) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        return res_json
+
+    def secretsmanager_http_put_pending_secret_value(
+        self, secret_id: str, secret_string: str
+    ) -> requests.Response:
+        http_body: json = {
+            "SecretId": secret_id,
+            "SecretString": secret_string,
+            "VersionStages": ["AWSPENDING"],
+        }
+        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_put_pending_secret_value_val_res(
+        res: requests.Response, secret_name: str
+    ) -> json:
+        return TestSecretsManager.secretsmanager_http_put_secret_value_val_res(res, secret_name)
+
+    def secretsmanager_http_put_secret_value_with(
+        self, secret_id: str, secret_string: str, client_request_token: Optional[str]
+    ) -> requests.Response:
+        http_body: json = {
+            "SecretId": secret_id,
+            "SecretString": secret_string,
+            "ClientRequestToken": client_request_token,
+        }
+        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_put_secret_value_with_val_res(
+        res: requests.Response, secret_name: str, client_request_token: str
+    ) -> json:
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        assert res_json["VersionId"] == client_request_token
+        return res_json
+
+    def secretsmanager_http_update_secret(
+        self, secret_id: str, secret_string: str, client_request_token: Optional[str]
+    ):
+        http_body: json = {"SecretId": secret_id, "SecretString": secret_string}
+        if client_request_token:
+            http_body["ClientRequestToken"] = client_request_token
+        return self.secretsmanager_http_json_post("secretsmanager.UpdateSecret", http_body)
+
+    @staticmethod
+    def secretsmanager_http_update_secret_val_res(
+        res: requests.Response, secret_name: str, client_request_token: Optional[str]
+    ):
+        assert res.status_code == 200
+        res_json: json = res.json()
+        assert res_json["Name"] == secret_name
+        if client_request_token:
+            assert res_json["VersionId"] == client_request_token
+        return res_json
+
+    def secretsmanager_http_put_secret_value_with_version(
+        self,
+        secret_id: str,
+        secret_string: str,
+        client_request_token: Optional[str],
+        version_stages: List[str],
+    ) -> requests.Response:
+        http_body: json = {
+            "SecretId": secret_id,
+            "SecretString": secret_string,
+            "ClientRequestToken": client_request_token,
+            "VersionStages": version_stages,
+        }
+        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
+
+    @staticmethod
+    def secretsmanager_http_put_secret_value_with_version_val_res(
+        res: requests.Response,
+        secret_name: str,
+        client_request_token: Optional[str],
+        version_stages: List[str],
+    ) -> json:
+        req_version_id: str
+        if client_request_token is None:
+            assert res.status_code == 200
+            req_version_id = res.json()["VersionId"]
+        else:
+            req_version_id = client_request_token
+        res_json = TestSecretsManager.secretsmanager_http_put_secret_value_with_val_res(
+            res, secret_name, req_version_id
+        )
+        assert res_json["VersionStages"] == version_stages
+        return res_json
+
+    def test_http_update_secret_with_missing_client_request_token(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "secret_string_v0"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        version_id_v0 = cr_v0_res_json["VersionId"]
+
+        # Update with client request token.
+        secret_string_v1: str = "secret_string_v1"
+        version_id_v1: str = str(uuid.uuid4())
+        self.secretsmanager_http_update_secret_val_res(
+            self.secretsmanager_http_update_secret(secret_name, secret_string_v1, version_id_v1),
+            secret_name,
+            version_id_v1,
+        )
+
+        # Get.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v1,
+            version_id_v1,
+        )
+
+        # Update without client request token.
+        secret_string_v2: str = "secret_string_v2"
+        res_update_json = self.secretsmanager_http_update_secret_val_res(
+            self.secretsmanager_http_update_secret(secret_name, secret_string_v2, None),
+            secret_name,
+            None,
+        )
+        version_id_v2 = res_update_json["VersionId"]
+        assert version_id_v2 != version_id_v1
+        assert version_id_v2 != version_id_v0
+
+        # Get.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v2,
+            version_id_v2,
+        )
+
+    def test_http_put_secret_value_with_new_custom_client_request_token(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with predefined ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        #
+        crt_v1: str = str(uuid.uuid4())
+        while crt_v1 == cr_v0_res_json["VersionId"]:
+            crt_v1 = str(uuid.uuid4())
+        #
+        self.secretsmanager_http_put_secret_value_val_res(
+            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, crt_v1),
+            secret_name,
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v1,
+            crt_v1,
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
+            {"VersionId": crt_v1, "VersionStages": ["AWSCURRENT"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_with_duplicate_client_request_token(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with duplicate ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        #
+        crt_v1: str = cr_v0_res_json["VersionId"]
+        #
+        self.secretsmanager_http_put_secret_value_val_res(
+            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, crt_v1),
+            secret_name,
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v1,
+            crt_v1,
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [{"VersionId": crt_v1, "VersionStages": ["AWSCURRENT"]}]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_with_null_client_request_token(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with null ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        #
+        pv_v1_res_json = self.secretsmanager_http_put_secret_value_val_res(
+            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, None),
+            secret_name,
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v1,
+            pv_v1_res_json["VersionId"],
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
+            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_with_undefined_client_request_token(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with undefined ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        #
+        pv_v1_res_json = self.secretsmanager_http_put_secret_value_val_res(
+            self.secretsmanager_http_put_secret_value(secret_name, secret_string_v1), secret_name
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v1,
+            pv_v1_res_json["VersionId"],
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
+            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_duplicate_req(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Duplicate update.
+        self.secretsmanager_http_put_secret_value_val_res(
+            self.secretsmanager_http_put_secret_value_with(
+                secret_name, secret_string_v0, cr_v0_res_json["VersionId"]
+            ),
+            secret_name,
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_null_client_request_token_new_version_stages(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with null ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        version_stages_v1: List[str] = ["AWSPENDING"]
+        #
+        pv_v1_res_json = self.secretsmanager_http_put_secret_value_with_version_val_res(
+            self.secretsmanager_http_put_secret_value_with_version(
+                secret_name, secret_string_v1, None, version_stages_v1
+            ),
+            secret_name,
+            None,
+            version_stages_v1,
+        )
+        #
+        assert pv_v1_res_json["VersionId"] != cr_v0_res_json["VersionId"]
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_with_val_res(
+            self.secretsmanager_http_get_secret_value_with(secret_name, "AWSPENDING"),
+            secret_name,
+            secret_string_v1,
+            pv_v1_res_json["VersionId"],
+            "AWSPENDING",
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
+            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSPENDING"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
+
+    def test_http_put_secret_value_custom_client_request_token_new_version_stages(self):
+        secret_name = f"s-{short_uid()}"
+
+        # Create v0.
+        secret_string_v0: str = "MySecretString"
+        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
+            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
+            secret_name,
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+
+        # Update v0 with null ClientRequestToken.
+        secret_string_v1: str = "MyNewSecretString"
+        version_stages_v1: List[str] = ["AWSPENDING"]
+        crt_v1: str = str(uuid.uuid4())
+        while crt_v1 == cr_v0_res_json["VersionId"]:
+            crt_v1 = str(uuid.uuid4())
+        #
+        self.secretsmanager_http_put_secret_value_with_version_val_res(
+            self.secretsmanager_http_put_secret_value_with_version(
+                secret_name, secret_string_v1, crt_v1, version_stages_v1
+            ),
+            secret_name,
+            crt_v1,
+            version_stages_v1,
+        )
+        #
+        # Check v1 base consistency.
+        self.secretsmanager_http_get_secret_value_with_val_res(
+            self.secretsmanager_http_get_secret_value_with(secret_name, "AWSPENDING"),
+            secret_name,
+            secret_string_v1,
+            crt_v1,
+            "AWSPENDING",
+        )
+        #
+        # Check v0 base consistency.
+        self.secretsmanager_http_get_secret_value_val_res(
+            self.secretsmanager_http_get_secret_value(secret_name),
+            secret_name,
+            secret_string_v0,
+            cr_v0_res_json["VersionId"],
+        )
+        #
+        # Check versioning base consistency.
+        versions_v0_v1: json = [
+            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
+            {"VersionId": crt_v1, "VersionStages": ["AWSPENDING"]},
+        ]
+        self.secretsmanager_http_list_secret_version_ids_val_res(
+            self.secretsmanager_http_list_secret_version_ids(secret_name),
+            secret_name,
+            versions_v0_v1,
+        )
+
+        self.secretsmanager_http_delete_secret_val_res(
+            self.secretsmanager_http_delete_secret(secret_name), secret_name
+        )
diff --git a/tests/integration/test_secretsmanager.py b/tests/integration/test_secretsmanager.py
deleted file mode 100644
index b6d00acf27846..0000000000000
--- a/tests/integration/test_secretsmanager.py
+++ /dev/null
@@ -1,830 +0,0 @@
-import json
-import uuid
-from datetime import datetime
-from typing import Dict, List, Optional
-
-import pytest
-import requests
-
-from localstack.constants import TEST_AWS_ACCOUNT_ID
-from localstack.services.awslambda.lambda_utils import LAMBDA_RUNTIME_PYTHON36
-from localstack.utils import testutil
-from localstack.utils.aws import aws_stack
-from localstack.utils.strings import short_uid
-from tests.integration.awslambda.test_lambda import TEST_LAMBDA_PYTHON_VERSION
-
-RESOURCE_POLICY = {
-    "Version": "2012-10-17",
-    "Statement": [
-        {
-            "Effect": "Allow",
-            "Principal": {"AWS": "arn:aws:iam::%s:root" % TEST_AWS_ACCOUNT_ID},
-            "Action": "secretsmanager:GetSecretValue",
-            "Resource": "*",
-        }
-    ],
-}
-
-
-class TestSecretsManager:
-    @pytest.fixture
-    def secretsmanager_client(self):
-        return aws_stack.create_external_boto_client("secretsmanager")
-
-    def test_create_and_update_secret(self, secretsmanager_client):
-        secret_name = "s-%s" % short_uid()
-        rs = secretsmanager_client.create_secret(
-            Name=secret_name,
-            SecretString="my_secret",
-            Description="testing creation of secrets",
-        )
-        secret_arn = rs["ARN"]
-
-        assert len(secret_arn.rpartition("-")[2]) == 6
-
-        rs = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs["Name"] == secret_name
-        assert rs["SecretString"] == "my_secret"
-        assert rs["ARN"] == secret_arn
-        assert isinstance(rs["CreatedDate"], datetime)
-
-        rs = secretsmanager_client.get_secret_value(SecretId=secret_arn)
-        assert rs["Name"] == secret_name
-        assert rs["SecretString"] == "my_secret"
-        assert rs["ARN"] == secret_arn
-
-        rs = secretsmanager_client.get_secret_value(SecretId=secret_arn[: len(secret_arn) - 6])
-        assert rs["Name"] == secret_name
-        assert rs["SecretString"] == "my_secret"
-        assert rs["ARN"] == secret_arn
-
-        rs = secretsmanager_client.get_secret_value(SecretId=secret_arn[: len(secret_arn) - 7])
-        assert rs["Name"] == secret_name
-        assert rs["SecretString"] == "my_secret"
-        assert rs["ARN"] == secret_arn
-
-        secretsmanager_client.put_secret_value(SecretId=secret_name, SecretString="new_secret")
-
-        rs = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs["Name"] == secret_name
-        assert rs["SecretString"] == "new_secret"
-
-        # update secret by ARN
-        rs = secretsmanager_client.update_secret(
-            SecretId=secret_arn, KmsKeyId="test123", Description="d1"
-        )
-        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 200
-        assert rs["ARN"] == secret_arn
-
-        # clean up
-        secretsmanager_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
-
-    def test_call_lists_secrets_multiple_time(self, secretsmanager_client):
-        secret_name = "s-%s" % short_uid()
-        secretsmanager_client.create_secret(
-            Name=secret_name,
-            SecretString="my_secret",
-            Description="testing creation of secrets",
-        )
-
-        # call list_secrets multiple times
-        for i in range(3):
-            rs = secretsmanager_client.list_secrets()
-            secrets = [secret for secret in rs["SecretList"] if secret["Name"] == secret_name]
-            assert 1 == len(secrets)
-
-        # clean up
-        secretsmanager_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
-
-    def test_create_multi_secrets(self, secretsmanager_client):
-        secret_names = [short_uid(), short_uid(), short_uid()]
-        arns = []
-        for secret_name in secret_names:
-            rs = secretsmanager_client.create_secret(
-                Name=secret_name,
-                SecretString="my_secret_{}".format(secret_name),
-                Description="testing creation of secrets",
-            )
-            arns.append(rs["ARN"])
-
-        rs = secretsmanager_client.list_secrets()
-        secrets = {
-            secret["Name"]: secret["ARN"]
-            for secret in rs["SecretList"]
-            if secret["Name"] in secret_names
-        }
-
-        assert len(secrets.keys()) == len(secret_names)
-        for arn in arns:
-            assert arn in secrets.values()
-
-        # clean up
-        for secret_name in secret_names:
-            secretsmanager_client.delete_secret(
-                SecretId=secret_name, ForceDeleteWithoutRecovery=True
-            )
-
-    def test_get_random_exclude_characters_and_symbols(self, secretsmanager_client):
-        random_password = secretsmanager_client.get_random_password(
-            PasswordLength=120, ExcludeCharacters="xyzDje@?!."
-        )
-
-        assert len(random_password["RandomPassword"]) == 120
-        assert all(c not in "xyzDje@?!." for c in random_password["RandomPassword"])
-
-    def test_resource_policy(self, secretsmanager_client):
-        secret_name = "s-%s" % short_uid()
-
-        secretsmanager_client.create_secret(
-            Name=secret_name,
-            SecretString="my_secret",
-            Description="testing creation of secrets",
-        )
-
-        secretsmanager_client.put_resource_policy(
-            SecretId=secret_name, ResourcePolicy=json.dumps(RESOURCE_POLICY)
-        )
-
-        rs = secretsmanager_client.get_resource_policy(SecretId=secret_name)
-
-        policy = json.loads(rs["ResourcePolicy"])
-
-        assert policy["Version"] == RESOURCE_POLICY["Version"]
-        assert policy["Statement"] == RESOURCE_POLICY["Statement"]
-
-        rs = secretsmanager_client.delete_resource_policy(SecretId=secret_name)
-        assert rs["ResponseMetadata"]["HTTPStatusCode"] == 200
-
-        # clean up
-        secretsmanager_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
-
-    def test_rotate_secret_with_lambda(self, secretsmanager_client):
-        secret_name = "s-%s" % short_uid()
-
-        secretsmanager_client.create_secret(
-            Name=secret_name,
-            SecretString="my_secret",
-            Description="testing rotation of secrets",
-        )
-
-        function_name = "s-%s" % short_uid()
-        function_arn = testutil.create_lambda_function(
-            handler_file=TEST_LAMBDA_PYTHON_VERSION,
-            func_name=function_name,
-            runtime=LAMBDA_RUNTIME_PYTHON36,
-        )["CreateFunctionResponse"]["FunctionArn"]
-
-        response = secretsmanager_client.rotate_secret(
-            SecretId=secret_name,
-            RotationLambdaARN=function_arn,
-            RotationRules={
-                "AutomaticallyAfterDays": 1,
-            },
-            RotateImmediately=True,
-        )
-
-        assert response["ResponseMetadata"]["HTTPStatusCode"] == 200
-
-        # clean up
-        secretsmanager_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
-        testutil.delete_lambda_function(function_name)
-
-    def test_put_secret_value_with_version_stages(self, secretsmanager_client):
-        secret_name: str = "s-%s" % short_uid()
-
-        secret_string_v0: str = "secret_string_v0"
-        cr_v0_res = secretsmanager_client.create_secret(
-            Name=secret_name, SecretString=secret_string_v0
-        )
-        pv_v0_vid: str = cr_v0_res["VersionId"]
-
-        rs_get_curr = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs_get_curr["SecretString"] == secret_string_v0
-        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
-
-        secret_string_v1: str = "secret_string_v1"
-        version_stages_v1: ["str"] = ["SAMPLESTAGE1", "SAMPLESTAGE0"]
-        pv_v1_vid: str = str(uuid.uuid4())
-        pv_v1_res = secretsmanager_client.put_secret_value(
-            SecretId=secret_name,
-            SecretString=secret_string_v1,
-            VersionStages=version_stages_v1,
-            ClientRequestToken=pv_v1_vid,
-        )
-        assert pv_v1_res["VersionId"] == pv_v1_vid
-        assert pv_v1_res["VersionStages"] == version_stages_v1
-
-        rs_get_curr = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs_get_curr["VersionId"] == pv_v0_vid
-        assert rs_get_curr["SecretString"] == secret_string_v0
-        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
-
-        secret_string_v2: str = "secret_string_v2"
-        version_stages_v2: ["str"] = version_stages_v1
-        pv_v2_vid: str = str(uuid.uuid4())
-        pv_v2_res = secretsmanager_client.put_secret_value(
-            SecretId=secret_name,
-            SecretString=secret_string_v2,
-            VersionStages=version_stages_v2,
-            ClientRequestToken=pv_v2_vid,
-        )
-        assert pv_v2_res["VersionId"] == pv_v2_vid
-        assert pv_v2_res["VersionStages"] == version_stages_v2
-
-        rs_get_curr = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs_get_curr["VersionId"] == pv_v0_vid
-        assert rs_get_curr["SecretString"] == secret_string_v0
-        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
-
-        secret_string_v3: str = "secret_string_v3"
-        version_stages_v3: ["str"] = ["AWSPENDING"]
-        pv_v3_vid: str = str(uuid.uuid4())
-        pv_v3_res = secretsmanager_client.put_secret_value(
-            SecretId=secret_name,
-            SecretString=secret_string_v3,
-            VersionStages=version_stages_v3,
-            ClientRequestToken=pv_v3_vid,
-        )
-        assert pv_v3_res["VersionId"] == pv_v3_vid
-        assert pv_v3_res["VersionStages"] == version_stages_v3
-
-        rs_get_curr = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs_get_curr["VersionId"] == pv_v0_vid
-        assert rs_get_curr["SecretString"] == secret_string_v0
-        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
-
-        secret_string_v4: str = "secret_string_v4"
-        pv_v4_vid: str = str(uuid.uuid4())
-        pv_v4_res = secretsmanager_client.put_secret_value(
-            SecretId=secret_name, SecretString=secret_string_v4, ClientRequestToken=pv_v4_vid
-        )
-        assert pv_v4_res["VersionId"] == pv_v4_vid
-        assert pv_v4_res["VersionStages"] == ["AWSCURRENT"]
-
-        rs_get_curr = secretsmanager_client.get_secret_value(SecretId=secret_name)
-        assert rs_get_curr["VersionId"] == pv_v4_vid
-        assert rs_get_curr["SecretString"] == secret_string_v4
-        assert rs_get_curr["VersionStages"] == ["AWSCURRENT"]
-
-        secretsmanager_client.delete_secret(SecretId=secret_name, ForceDeleteWithoutRecovery=True)
-
-    @staticmethod
-    def secretsmanager_http_json_headers(amz_target: str) -> Dict:
-        headers = aws_stack.mock_aws_request_headers("secretsmanager")
-        headers["X-Amz-Target"] = amz_target
-        return headers
-
-    def secretsmanager_http_json_post(self, amz_target: str, http_body: json) -> requests.Response:
-        ep_url: str = aws_stack.get_local_service_url("secretsmanager")
-        http_headers: Dict = self.secretsmanager_http_json_headers(amz_target)
-        return requests.post(ep_url, headers=http_headers, data=json.dumps(http_body))
-
-    def secretsmanager_http_create_secret_string(
-        self, secret_name: str, secret_string: str
-    ) -> requests.Response:
-        http_body: json = {"Name": secret_name, "SecretString": secret_string}
-        return self.secretsmanager_http_json_post("secretsmanager.CreateSecret", http_body)
-
-    @staticmethod
-    def secretsmanager_http_create_secret_string_val_res(
-        res: requests.Response, secret_name: str
-    ) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_name
-        return res_json
-
-    def secretsmanager_http_delete_secret(self, secret_id: str) -> requests.Response:
-        http_body: json = {"SecretId": secret_id}
-        return self.secretsmanager_http_json_post("secretsmanager.DeleteSecret", http_body)
-
-    @staticmethod
-    def secretsmanager_http_delete_secret_val_res(res: requests.Response, secret_id: str) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_id
-        return res_json
-
-    def secretsmanager_http_get_secret_value(self, secret_id: str) -> requests.Response:
-        http_body: json = {"SecretId": secret_id}
-        return self.secretsmanager_http_json_post("secretsmanager.GetSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_get_secret_value_val_res(
-        res: requests.Response, secret_name: str, secret_string: str, version_id: str
-    ) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_name
-        assert res_json["SecretString"] == secret_string
-        assert res_json["VersionId"] == version_id
-        return res_json
-
-    def secretsmanager_http_get_secret_value_with(
-        self, secret_id: str, version_stage: str
-    ) -> requests.Response:
-        http_body: json = {"SecretId": secret_id, "VersionStage": version_stage}
-        return self.secretsmanager_http_json_post("secretsmanager.GetSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_get_secret_value_with_val_res(
-        res: requests.Response,
-        secret_name: str,
-        secret_string: str,
-        version_id: str,
-        version_stage: str,
-    ) -> json:
-        res_json = TestSecretsManager.secretsmanager_http_get_secret_value_val_res(
-            res, secret_name, secret_string, version_id
-        )
-        assert res_json["VersionStages"] == [version_stage]
-        return res_json
-
-    def secretsmanager_http_list_secret_version_ids(self, secret_id: str) -> requests.Response:
-        http_body: json = {"SecretId": secret_id}
-        return self.secretsmanager_http_json_post("secretsmanager.ListSecretVersionIds", http_body)
-
-    @staticmethod
-    def secretsmanager_http_list_secret_version_ids_val_res(
-        res: requests.Response, secret_name: str, versions: json
-    ) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_name
-        res_versions: [json] = res_json["Versions"]
-        assert len(res_versions) == len(versions)
-        assert len(set([rv["VersionId"] for rv in res_versions])) == len(res_versions)
-        assert len(set([v["VersionId"] for v in versions])) == len(versions)
-        for version in versions:
-            vs_in_res: [json] = list(
-                filter(lambda rv: rv["VersionId"] == version["VersionId"], res_versions)
-            )
-            assert len(vs_in_res) == 1
-            v_in_res = vs_in_res[0]
-            assert v_in_res["VersionStages"] == version["VersionStages"]
-        return res_json
-
-    def secretsmanager_http_put_secret_value(
-        self, secret_id: str, secret_string: str
-    ) -> requests.Response:
-        http_body: json = {
-            "SecretId": secret_id,
-            "SecretString": secret_string,
-        }
-        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_put_secret_value_val_res(
-        res: requests.Response, secret_name: str
-    ) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_name
-        return res_json
-
-    def secretsmanager_http_put_pending_secret_value(
-        self, secret_id: str, secret_string: str
-    ) -> requests.Response:
-        http_body: json = {
-            "SecretId": secret_id,
-            "SecretString": secret_string,
-            "VersionStages": ["AWSPENDING"],
-        }
-        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_put_pending_secret_value_val_res(
-        res: requests.Response, secret_name: str
-    ) -> json:
-        return TestSecretsManager.secretsmanager_http_put_secret_value_val_res(res, secret_name)
-
-    def secretsmanager_http_put_secret_value_with(
-        self, secret_id: str, secret_string: str, client_request_token: Optional[str]
-    ) -> requests.Response:
-        http_body: json = {
-            "SecretId": secret_id,
-            "SecretString": secret_string,
-            "ClientRequestToken": client_request_token,
-        }
-        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_put_secret_value_with_val_res(
-        res: requests.Response, secret_name: str, client_request_token: str
-    ) -> json:
-        assert res.status_code == 200
-        res_json: json = res.json()
-        assert res_json["Name"] == secret_name
-        assert res_json["VersionId"] == client_request_token
-        return res_json
-
-    def secretsmanager_http_put_secret_value_with_version(
-        self,
-        secret_id: str,
-        secret_string: str,
-        client_request_token: Optional[str],
-        version_stages: List[str],
-    ) -> requests.Response:
-        http_body: json = {
-            "SecretId": secret_id,
-            "SecretString": secret_string,
-            "ClientRequestToken": client_request_token,
-            "VersionStages": version_stages,
-        }
-        return self.secretsmanager_http_json_post("secretsmanager.PutSecretValue", http_body)
-
-    @staticmethod
-    def secretsmanager_http_put_secret_value_with_version_val_res(
-        res: requests.Response,
-        secret_name: str,
-        client_request_token: Optional[str],
-        version_stages: List[str],
-    ) -> json:
-        req_version_id: str
-        if client_request_token is None:
-            assert res.status_code == 200
-            req_version_id = res.json()["VersionId"]
-        else:
-            req_version_id = client_request_token
-        res_json = TestSecretsManager.secretsmanager_http_put_secret_value_with_val_res(
-            res, secret_name, req_version_id
-        )
-        assert res_json["VersionStages"] == version_stages
-        return res_json
-
-    def test_http_put_secret_value_with_new_custom_client_request_token(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with predefined ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        #
-        crt_v1: str = str(uuid.uuid4())
-        while crt_v1 == cr_v0_res_json["VersionId"]:
-            crt_v1 = str(uuid.uuid4())
-        #
-        self.secretsmanager_http_put_secret_value_val_res(
-            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, crt_v1),
-            secret_name,
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v1,
-            crt_v1,
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
-            {"VersionId": crt_v1, "VersionStages": ["AWSCURRENT"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_with_duplicate_client_request_token(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with duplicate ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        #
-        crt_v1: str = cr_v0_res_json["VersionId"]
-        #
-        self.secretsmanager_http_put_secret_value_val_res(
-            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, crt_v1),
-            secret_name,
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v1,
-            crt_v1,
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [{"VersionId": crt_v1, "VersionStages": ["AWSCURRENT"]}]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_with_null_client_request_token(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with null ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        #
-        pv_v1_res_json = self.secretsmanager_http_put_secret_value_val_res(
-            self.secretsmanager_http_put_secret_value_with(secret_name, secret_string_v1, None),
-            secret_name,
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v1,
-            pv_v1_res_json["VersionId"],
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
-            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_with_undefined_client_request_token(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with undefined ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        #
-        pv_v1_res_json = self.secretsmanager_http_put_secret_value_val_res(
-            self.secretsmanager_http_put_secret_value(secret_name, secret_string_v1), secret_name
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v1,
-            pv_v1_res_json["VersionId"],
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSPREVIOUS"]},
-            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_duplicate_req(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Duplicate update.
-        self.secretsmanager_http_put_secret_value_val_res(
-            self.secretsmanager_http_put_secret_value_with(
-                secret_name, secret_string_v0, cr_v0_res_json["VersionId"]
-            ),
-            secret_name,
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_null_client_request_token_new_version_stages(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with null ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        version_stages_v1: List[str] = ["AWSPENDING"]
-        #
-        pv_v1_res_json = self.secretsmanager_http_put_secret_value_with_version_val_res(
-            self.secretsmanager_http_put_secret_value_with_version(
-                secret_name, secret_string_v1, None, version_stages_v1
-            ),
-            secret_name,
-            None,
-            version_stages_v1,
-        )
-        #
-        assert pv_v1_res_json["VersionId"] != cr_v0_res_json["VersionId"]
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_with_val_res(
-            self.secretsmanager_http_get_secret_value_with(secret_name, "AWSPENDING"),
-            secret_name,
-            secret_string_v1,
-            pv_v1_res_json["VersionId"],
-            "AWSPENDING",
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
-            {"VersionId": pv_v1_res_json["VersionId"], "VersionStages": ["AWSPENDING"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )
-
-    def test_http_put_secret_value_custom_client_request_token_new_version_stages(self):
-        secret_name: str = "s-%s" % short_uid()
-
-        # Create v0.
-        secret_string_v0: str = "MySecretString"
-        cr_v0_res_json: json = self.secretsmanager_http_create_secret_string_val_res(
-            self.secretsmanager_http_create_secret_string(secret_name, secret_string_v0),
-            secret_name,
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-
-        # Update v0 with null ClientRequestToken.
-        secret_string_v1: str = "MyNewSecretString"
-        version_stages_v1: List[str] = ["AWSPENDING"]
-        crt_v1: str = str(uuid.uuid4())
-        while crt_v1 == cr_v0_res_json["VersionId"]:
-            crt_v1 = str(uuid.uuid4())
-        #
-        self.secretsmanager_http_put_secret_value_with_version_val_res(
-            self.secretsmanager_http_put_secret_value_with_version(
-                secret_name, secret_string_v1, crt_v1, version_stages_v1
-            ),
-            secret_name,
-            crt_v1,
-            version_stages_v1,
-        )
-        #
-        # Check v1 base consistency.
-        self.secretsmanager_http_get_secret_value_with_val_res(
-            self.secretsmanager_http_get_secret_value_with(secret_name, "AWSPENDING"),
-            secret_name,
-            secret_string_v1,
-            crt_v1,
-            "AWSPENDING",
-        )
-        #
-        # Check v0 base consistency.
-        self.secretsmanager_http_get_secret_value_val_res(
-            self.secretsmanager_http_get_secret_value(secret_name),
-            secret_name,
-            secret_string_v0,
-            cr_v0_res_json["VersionId"],
-        )
-        #
-        # Check versioning base consistency.
-        versions_v0_v1: json = [
-            {"VersionId": cr_v0_res_json["VersionId"], "VersionStages": ["AWSCURRENT"]},
-            {"VersionId": crt_v1, "VersionStages": ["AWSPENDING"]},
-        ]
-        self.secretsmanager_http_list_secret_version_ids_val_res(
-            self.secretsmanager_http_list_secret_version_ids(secret_name),
-            secret_name,
-            versions_v0_v1,
-        )
-
-        self.secretsmanager_http_delete_secret_val_res(
-            self.secretsmanager_http_delete_secret(secret_name), secret_name
-        )


Code review comments:
diff_hunk: @@ -68,143 +84,458 @@ def __init__(self):
         apply_patches()
 
     @staticmethod
-    def _transform_context_secret_id(context: RequestContext) -> Optional[Dict]:
+    def _transform_context_secret_id(secret_id: SecretIdType) -> Optional[SecretIdType]:
         # If secret ARN ends with "-<randomId>" this is removed from the request for upstream compatibility.
-        data = json.loads(to_str(context.request.data or "{}"))
-        secret_id = data.get("SecretId", None)
-        if secret_id and ":" in secret_id:
+        t_secret_id = secret_id
+        if secret_id and re.match(r"^arn:", secret_id):
             arn = aws_stack.parse_arn(secret_id)
             aws_region = aws_stack.get_region()
             if arn["region"] != aws_region:
                 LOG.info(f'Expected request region "{aws_region}" for secret "{secret_id}"')
             resource_id = arn["resource"].split(":")[-1]
             if resource_id[-7] == "-":
-                data["SecretId"] = resource_id[:-7]
+                t_secret_id = resource_id[:-7]
             elif resource_id[-1] != "-":
-                data["SecretId"] += "-"
-            return data
-        return None
+                t_secret_id += "-"
+        return None if secret_id == t_secret_id else t_secret_id
+
+    @staticmethod
+    def _validate_secret_id(secret_id: SecretIdType) -> bool:
+        # The secret name can contain ASCII letters, numbers, and the following characters: /_+=.@-
+        return bool(re.match(r"^[A-Za-z0-9/_+=.@-]+\Z", secret_id))
+
+    @staticmethod
+    def _raise_if_invalid_secret_id(secret_id: Union[SecretIdType, NameType]):
+        # Patches moto's implementation for which secret_ids are not validated, by raising a ValidationException.
+        # Skips this check if the secret_id provided appears to be an arn (starting with 'arn:').
+        if not re.match(
+            r"^arn:", secret_id
+        ):  # Check if it appears to be an arn: so to skip secret_id check: delegate parsing of arn to handlers.
+            if not SecretsmanagerProvider._validate_secret_id(secret_id):
+                raise ValidationException(
+                    "Invalid name. Must be a valid name containing alphanumeric "
+                    "characters, or any of the following: -/_+=.@!"
+                )
 
     @staticmethod
-    def _call_moto_with_request_secret_id(context: RequestContext) -> ServiceResponse:
-        data_dict = SecretsmanagerProvider._transform_context_secret_id(context)
-        return call_moto_with_request(context, data_dict) if data_dict else call_moto(context)
+    def _call_moto_with_request_secret_id(
+        context: RequestContext, request: Dict
+    ) -> ServiceResponse:
+        t_secret_id = SecretsmanagerProvider._transform_context_secret_id(
+            request.get("SecretId", None)
+        )
+        if t_secret_id:
+            request.update({"SecretId": t_secret_id})
+            return call_moto_with_request(context, request)
+        return call_moto(context)
 
     @handler("CancelRotateSecret", expand=False)
     def cancel_rotate_secret(
         self, context: RequestContext, request: CancelRotateSecretRequest
     ) -> CancelRotateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("CreateSecret", expand=False)
     def create_secret(
         self, context: RequestContext, request: CreateSecretRequest
     ) -> CreateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["Name"])
+        return call_moto(context)
 
     @handler("DeleteResourcePolicy", expand=False)
     def delete_resource_policy(
         self, context: RequestContext, request: DeleteResourcePolicyRequest
     ) -> DeleteResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("DeleteSecret", expand=False)
     def delete_secret(
         self, context: RequestContext, request: DeleteSecretRequest
     ) -> DeleteSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("DescribeSecret", expand=False)
     def describe_secret(
         self, context: RequestContext, request: DescribeSecretRequest
     ) -> DescribeSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        res = self._call_moto_with_request_secret_id(context, request)
+        return res
 
     @handler("GetResourcePolicy", expand=False)
     def get_resource_policy(
         self, context: RequestContext, request: GetResourcePolicyRequest
     ) -> GetResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("GetSecretValue", expand=False)
     def get_secret_value(
         self, context: RequestContext, request: GetSecretValueRequest
     ) -> GetSecretValueResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ListSecretVersionIds", expand=False)
     def list_secret_version_ids(
         self, context: RequestContext, request: ListSecretVersionIdsRequest
     ) -> ListSecretVersionIdsResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("PutResourcePolicy", expand=False)
     def put_resource_policy(
         self, context: RequestContext, request: PutResourcePolicyRequest
     ) -> PutResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("PutSecretValue", expand=False)
     def put_secret_value(
         self, context: RequestContext, request: PutSecretValueRequest
     ) -> PutSecretValueResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RemoveRegionsFromReplication", expand=False)
     def remove_regions_from_replication(
         self, context: RequestContext, request: RemoveRegionsFromReplicationRequest
     ) -> RemoveRegionsFromReplicationResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ReplicateSecretToRegions", expand=False)
     def replicate_secret_to_regions(
         self, context: RequestContext, request: ReplicateSecretToRegionsRequest
     ) -> ReplicateSecretToRegionsResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RestoreSecret", expand=False)
     def restore_secret(
         self, context: RequestContext, request: RestoreSecretRequest
     ) -> RestoreSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("RotateSecret", expand=False)
     def rotate_secret(
         self, context: RequestContext, request: RotateSecretRequest
     ) -> RotateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("StopReplicationToReplica", expand=False)
     def stop_replication_to_replica(
         self, context: RequestContext, request: StopReplicationToReplicaRequest
     ) -> StopReplicationToReplicaResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("TagResource", expand=False)
     def tag_resource(self, context: RequestContext, request: TagResourceRequest) -> None:
-        self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UntagResource", expand=False)
     def untag_resource(self, context: RequestContext, request: UntagResourceRequest) -> None:
-        self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UpdateSecret", expand=False)
     def update_secret(
         self, context: RequestContext, request: UpdateSecretRequest
     ) -> UpdateSecretResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("UpdateSecretVersionStage", expand=False)
     def update_secret_version_stage(
         self, context: RequestContext, request: UpdateSecretVersionStageRequest
     ) -> UpdateSecretVersionStageResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
 
     @handler("ValidateResourcePolicy", expand=False)
     def validate_resource_policy(
         self, context: RequestContext, request: ValidateResourcePolicyRequest
     ) -> ValidateResourcePolicyResponse:
-        return self._call_moto_with_request_secret_id(context)
+        self._raise_if_invalid_secret_id(request["SecretId"])
+        return self._call_moto_with_request_secret_id(context, request)
+
+
+@patch(FakeSecret.__init__)
+def fake_secret__init__(fn, self, **kwargs):
+    fn(self, **kwargs)
+
+    # Fix time not including millis.
+    time_now = time.time()
+    if kwargs.get("last_changed_date", None):
+        self.last_changed_date = time_now
+    if kwargs.get("created_date", None):
+        self.created_date = time_now
+
+    # The last date that the secret value was retrieved.
+    # This value does not include the time.
+    # This field is omitted if the secret has never been retrieved.
+    self.last_accessed_date = None
+
+
+@patch(FakeSecret.update)
+def fake_secret_update(
+    fn, self, description=None, tags=None, kms_key_id=None, last_changed_date=None
+):
+    fn(self, description, tags, kms_key_id, last_changed_date)
+    if last_changed_date is not None:
+        self.last_changed_date = time.time()
+
+
+class FakeSecretVersionStore(dict):
+    def __setitem__(self, key, value):
+        self.put_version(key, value, time.time())
+
+    def put_version(self, version_id: str, version: Dict, create_date: Optional[float] = None):
+        if create_date and "createdate" in version:
+            version["createdate"] = create_date
+        super().__setitem__(version_id, version)
+
+
+@patch(FakeSecret.set_versions)
+def fake_secret_set_versions(_, self, versions):
+    self.versions = FakeSecretVersionStore()
+    for version_id, version in versions.items():
+        self.versions.put_version(version_id, version, self.created_date)
+
+
+@patch(SecretsManagerBackend.get_secret_value)
+def moto_smb_get_secret_value(fn, self, secret_id, version_id, version_stage):
+    res = fn(self, secret_id, version_id, version_stage)
+
+    secret_id = self.secrets[secret_id]
+    if secret_id:  # Redundant, we know from the response it exists: no exceptions.
+        secret_id.last_accessed_date = today_no_time()
+    else:
+        LOG.warning(
+            f'Expected Secret to exist on non failing GetSecretValue request for SecretId "{secret_id}"'
+        )
+
+    return res
+
+
+@patch(FakeSecret.to_dict)
+def fake_secret_to_dict(fn, self):
+    res_dict = fn(self)
+    if self.last_accessed_date:
+        res_dict["LastAccessedDate"] = self.last_accessed_date
+    return res_dict
+
+
+@patch(SecretsManagerBackend.update_secret)
+def backend_update_secret(
+    fn,
+    self,
+    secret_id,
+    description=None,
+    **kwargs,
+):
+    fn(self, secret_id, **kwargs)
+    secret = self.secrets[secret_id]
+
+    # Fix missing update of secret description.
+    # Secret exists if this point is reached.
+    if description is not None:
+        secret.description = description
+
+    return secret.to_short_dict()
+
+
+@patch(SecretsManagerResponse.update_secret)
+def response_update_secret(_, self):
+    secret_id = self._get_param("SecretId")
+    description = self._get_param("Description")
+    secret_string = self._get_param("SecretString")
+    secret_binary = self._get_param("SecretBinary")
+    client_request_token = self._get_param("ClientRequestToken")
+    kms_key_id = self._get_param("KmsKeyId")
+    return secretsmanager_backends[self.region].update_secret(
+        secret_id=secret_id,
+        description=description,
+        secret_string=secret_string,
+        secret_binary=secret_binary,
+        client_request_token=client_request_token,
+        kms_key_id=kms_key_id,
+    )
+
+
+@patch(SecretsManagerBackend.update_secret_version_stage)
+def backend_update_secret_version_stage(
+    fn, self, secret_id, version_stage, remove_from_version_id, move_to_version_id
+):
+    fn(self, secret_id, version_stage, remove_from_version_id, move_to_version_id)
+
+    secret = self.secrets[secret_id]
+    if version_stage == AWSCURRENT:
+        secret.default_version_id = move_to_version_id
+
+        # Ensure only one AWSPREVIOUS tagged version is in the pool.
+        # Remove secret versions with no version stages.
+        versions_no_stages = []
+        update_vid_set = {remove_from_version_id, move_to_version_id}
+        for version_id, version in secret.versions.items():
+            version_stages = version["version_stages"]
+            if version_id not in update_vid_set and AWSPREVIOUS in version_stages:
+                version_stages.remove(AWSPREVIOUS)
+                if not version_stages:
+                    versions_no_stages.append(version_id)
+
+        for version_no_stages in versions_no_stages:
+            del secret.versions[version_no_stages]
+
+    return json.dumps({"ARN": secret.arn, "Name": secret.name})
+
+
+@patch(FakeSecret.reset_default_version)
+def fake_secret_reset_default_version(fn, self, secret_version, version_id):
+    fn(self, secret_version, version_id)
+    # Remove versions with no version stages.
+    versions_no_stages = [
+        version_id for version_id, version in self.versions.items() if not version["version_stages"]
+    ]
+    for version_no_stages in versions_no_stages:
+        del self.versions[version_no_stages]
+
+
+@patch(FakeSecret.remove_version_stages_from_old_versions)
+def fake_secret_remove_version_stages_from_old_versions(fn, self, version_stages):
+    fn(self, version_stages)
+    # Remove versions with no version stages.
+    versions_no_stages = [
+        version_id for version_id, version in self.versions.items() if not version["version_stages"]
+    ]
+    for version_no_stages in versions_no_stages:
+        del self.versions[version_no_stages]
+
+
+@patch(SecretsManagerBackend.rotate_secret)
+def backend_rotate_secret(
+    _,
+    self,
+    secret_id,
+    client_request_token=None,
+    rotation_lambda_arn=None,
+    rotation_rules=None,
+):
+    rotation_days = "AutomaticallyAfterDays"
+
+    if not self._is_valid_identifier(secret_id):
+        raise SecretNotFoundException()
+
+    if self.secrets[secret_id].is_deleted():
+        raise InvalidRequestException(
+            "An error occurred (InvalidRequestException) when calling the RotateSecret operation: You tried to \
+            perform the operation on a secret that's currently marked deleted."
+        )
+
+    if rotation_lambda_arn:
+        if len(rotation_lambda_arn) > 2048:
+            msg = "RotationLambdaARN " "must <= 2048 characters long."
+            raise InvalidParameterException(msg)
+
+    if rotation_rules:
+        if rotation_days in rotation_rules:
+            rotation_period = rotation_rules[rotation_days]
+            if rotation_period < 1 or rotation_period > 1000:
+                msg = "RotationRules.AutomaticallyAfterDays " "must be within 1-1000."
+                raise InvalidParameterException(msg)
+
+    secret = self.secrets[secret_id]
+
+    # The rotation function must end with the versions of the secret in
+    # one of two states:
+    #
+    #  - The AWSPENDING and AWSCURRENT staging labels are attached to the
+    #    same version of the secret, or
+    #  - The AWSPENDING staging label is not attached to any version of the secret.
+    #
+    # If the AWSPENDING staging label is present but not attached to the same
+    # version as AWSCURRENT then any later invocation of RotateSecret assumes
+    # that a previous rotation request is still in progress and returns an error.
+    try:
+        version = next(
+            version
+            for version in secret.versions.values()
+            if AWSPENDING in version["version_stages"]
+        )
+        if AWSCURRENT in version["version_stages"]:
+            msg = "Previous rotation request is still in progress."
+            raise InvalidRequestException(msg)
+
+    except StopIteration:
+        # Pending is not present in any version
+        pass
+
+    if client_request_token:
+        self._client_request_token_validator(client_request_token)
+        new_version_id = client_request_token
+    else:
+        new_version_id = str(uuid.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".
+    self._add_secret(
+        secret_id,
+        None,
+        description=secret.description,
+        tags=secret.tags,
+        version_id=new_version_id,
+        version_stages=[AWSPENDING],
+    )
+    secret.rotation_lambda_arn = rotation_lambda_arn or ""
+    if rotation_rules:
+        secret.auto_rotate_after_days = rotation_rules.get(rotation_days, 0)
+    if secret.auto_rotate_after_days > 0:
+        secret.rotation_enabled = True
+
+    # Begin the rotation process for the given secret by invoking the lambda function.
+    if secret.rotation_lambda_arn:
+        from moto.awslambda.models import lambda_backends
+
+        lambda_backend = lambda_backends[self.region]
+
+        request_headers = {}
+        response_headers = {}
+
+        func = lambda_backend.get_function(secret.rotation_lambda_arn)
+        if not func:
+            msg = "Resource not found for ARN '{}'.".format(secret.rotation_lambda_arn)
+            raise ResourceNotFoundException(msg)
+
+        for step in ["create", "set", "test", "finish"]:
+            func.invoke(
+                json.dumps(
+                    {
+                        "Step": step + "Secret",
+                        "SecretId": secret.name,
+                        "ClientRequestToken": new_version_id,
+                    }
+                ),
+                request_headers,
+                response_headers,
+            )
+
+        secret.set_default_version_id(new_version_id)
+        version_stages = secret.versions[new_version_id]["version_stages"]
+        if AWSPENDING in version_stages:
+            version_stages.remove(AWSPENDING)
path: localstack/services/secretsmanager/provider.py
text: @user1:
I feel like this section is similar to above where it's going beyond what AWS does, which can lead to unintended behavior.  While it doesn't impact me, it may be worth considering removing this so localstack can better mirror AWS behavior.

@author:
This seems to remove AWSPENDING from the now AWSCURRENT secret version once the rotation has terminated.