Repository: getmoto/moto
Pull request #9008 (2025-06-24 20:06:59)

S3: get_object() should not return ContentEncoding=aws-chunked
Fixes #9007 

## Background
`botocore` sometimes adds the Content-Encoding header `aws-chunked` (since https://github.com/boto/botocore/pull/2859)

Moto receives the headers, and returns it as-is. But: AWS: removes the `aws-chunked` part, and only returns whatever the user explicitly provided.

Note that in `botocore==1.35.0` this did not happen - the `aws-chunked` encoding is not added. My assumption is that this changed with https://github.com/boto/boto3/issues/4392 (fixed on our side with https://github.com/getmoto/moto/pull/8495), which would align with the observation from the user that this started around February. I haven't done any deepdive into the root cause though.

This PR explicitly removes the `aws-chunked` value (if provided), improving parity with AWS.


Patch:
diff --git a/moto/core/responses.py b/moto/core/responses.py
index be15dab19..e53bd5bae 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import datetime
 import functools
 import json
@@ -5,6 +7,7 @@ import logging
 import os
 import re
 from collections import OrderedDict, defaultdict
+from dataclasses import dataclass
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -24,13 +27,14 @@ from xml.dom.minidom import parseString as parseXML
 import boto3
 import requests
 import xmltodict
+from botocore.model import OperationModel, ServiceModel
 from jinja2 import DictLoader, Environment, Template
 from werkzeug.exceptions import HTTPException
 
 from moto import settings
 from moto.core.common_types import TYPE_IF_NONE, TYPE_RESPONSE
 from moto.core.exceptions import DryRunClientError, ServiceException
-from moto.core.serialize import SERIALIZERS, XFormedAttributePicker
+from moto.core.serialize import SERIALIZERS, ResponseSerializer, XFormedAttributePicker
 from moto.core.utils import (
     camelcase_to_underscores,
     get_service_model,
@@ -279,6 +283,14 @@ class ActionAuthenticatorMixin(object):
         return decorator
 
 
+@dataclass
+class ActionContext:
+    service_model: ServiceModel
+    operation_model: OperationModel
+    serializer_class: type[ResponseSerializer]
+    response_class: type[BaseResponse]
+
+
 class ActionResult:
     """Wrapper class for serializable results returned from `responses.py` methods."""
 
@@ -289,6 +301,34 @@ class ActionResult:
     def result(self) -> object:
         return self._result
 
+    def execute_result(self, context: ActionContext) -> TYPE_RESPONSE:
+        """
+        Execute the result in the context of the given service and operation model.
+        This is a placeholder for any logic that might be needed to process the result
+        based on the service and operation context.
+        """
+        serializer_cls = context.serializer_class
+        response_transformers = getattr(
+            context.response_class, "RESPONSE_KEY_PATH_TO_TRANSFORMER", None
+        )
+        value_picker = XFormedAttributePicker(
+            response_transformers=response_transformers
+        )
+        serializer = serializer_cls(
+            operation_model=context.operation_model,
+            pretty_print=settings.PRETTIFY_RESPONSES,
+            value_picker=value_picker,
+        )
+        serialized = serializer.serialize(self.result)
+        return serialized["status_code"], serialized["headers"], serialized["body"]  # type: ignore[return-value]
+
+
+class EmptyResult(ActionResult):
+    """A special ActionResult that represents an empty result."""
+
+    def __init__(self) -> None:
+        super().__init__(None)
+
 
 class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
     default_region = "us-east-1"
@@ -595,13 +635,10 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
         service_model = get_service_model(self.service_name)
         operation_model = service_model.operation_model(self._get_action())
         serializer_cls = SERIALIZERS[service_model.protocol]
-        serializer = serializer_cls(
-            operation_model=operation_model,
-            pretty_print=settings.PRETTIFY_RESPONSES,
-            value_picker=XFormedAttributePicker(),
+        context = ActionContext(
+            service_model, operation_model, serializer_cls, self.__class__
         )
-        serialized = serializer.serialize(action_result.result)
-        return serialized["status_code"], serialized["headers"], serialized["body"]  # type: ignore[return-value]
+        return action_result.execute_result(context)
 
     def call_action(self) -> TYPE_RESPONSE:
         headers = self.response_headers
diff --git a/moto/core/serialize.py b/moto/core/serialize.py
index a94c11e7e..59c9cc3f4 100644
--- a/moto/core/serialize.py
+++ b/moto/core/serialize.py
@@ -58,7 +58,16 @@ import calendar
 import json
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Generator, Mapping, MutableMapping, Optional, TypedDict, Union
+from typing import (
+    Any,
+    Callable,
+    Generator,
+    Mapping,
+    MutableMapping,
+    Optional,
+    TypedDict,
+    Union,
+)
 
 import xmltodict
 from botocore import xform_name
@@ -68,6 +77,7 @@ from botocore.model import (
     MapShape,
     NoShapeFoundError,
     OperationModel,
+    ServiceModel,
     Shape,
     StructureShape,
 )
@@ -255,6 +265,7 @@ class ResponseSerializer(ShapeHelpersMixin):
             value_picker = DefaultAttributePicker()
         self._value_picker = value_picker
         self._timestamp_serializer = TimestampSerializer(self.DEFAULT_TIMESTAMP_FORMAT)
+        self.path = []
 
     def _create_default_response(self) -> ResponseDict:
         response_dict: ResponseDict = {
@@ -334,7 +345,14 @@ class ResponseSerializer(ShapeHelpersMixin):
         return base64.b64encode(value).strip().decode(self.DEFAULT_ENCODING)
 
     def get_value(self, value: Any, key: str, shape: Shape) -> Any:
-        context = AttributePickerContext(obj=value, key=key, shape=shape)
+        context = AttributePickerContext(
+            obj=value,
+            key=key,
+            shape=shape,
+            operation_model=self.operation_model,
+            service_model=self.operation_model.service_model,
+            key_path=".".join(self.path),
+        )
         return self._value_picker(context)
 
     @staticmethod
@@ -377,7 +395,12 @@ class ResponseSerializer(ShapeHelpersMixin):
         method = getattr(
             self, "_serialize_type_%s" % shape.type_name, self._default_serialize
         )
+        if not key:
+            self.path.append(shape.name)
+        else:
+            self.path.append(key)
         method(serialized, value, shape, key)
+        self.path.pop()
 
     @staticmethod
     def _default_serialize(
@@ -555,9 +578,10 @@ class BaseJSONSerializer(ResponseSerializer):
             # a __current__ key on a wrapper dict to serialize each
             # list item before appending it to the serialized list.
             assert isinstance(shape.member, Shape)  # mypy hint
-            self._serialize(wrapper, list_item, shape.member, "__current__")
-            if "__current__" in wrapper:
-                list_obj.append(wrapper["__current__"])
+            item_key = shape.member.name
+            self._serialize(wrapper, list_item, shape.member, item_key)
+            if item_key in wrapper:
+                list_obj.append(wrapper[item_key])
             else:
                 list_obj.append(list_item)
 
@@ -677,9 +701,10 @@ class BaseXMLSerializer(ResponseSerializer):
             items_name = self.get_serialized_name(shape.member, "member")
             serialized[key] = {items_name: list_obj}
         for list_item in value:
+            item_key = shape.member.name
             wrapper = {}
-            self._serialize(wrapper, list_item, shape.member, "__current__")
-            list_obj.append(wrapper["__current__"])
+            self._serialize(wrapper, list_item, shape.member, item_key)
+            list_obj.append(wrapper[item_key])
         if not list_obj:
             serialized[key] = ""
 
@@ -908,6 +933,9 @@ class AttributePickerContext:
     obj: Any
     key: str
     shape: Shape
+    operation_model: OperationModel | None = None
+    service_model: ServiceModel | None = None
+    key_path: str = ""
 
 
 class DefaultAttributePicker:
@@ -915,15 +943,20 @@ class DefaultAttributePicker:
         return get_value(context.obj, context.key, None)
 
 
-class AliasedAttributePicker(DefaultAttributePicker):
+class AttributePicker(DefaultAttributePicker):
     """Uses alias providers to find the value of an attribute in a Python object"""
 
     def __init__(
-        self, alias_providers: list[type[AttributeAliasProvider]] | None = None
+        self,
+        alias_providers: list[type[AttributeAliasProvider]] | None = None,
+        response_transformers: dict[str, Callable[[Any], Any]] | None = None,
     ) -> None:
         self.alias_providers = (
             alias_providers if alias_providers is not None else DEFAULT_ALIAS_PROVIDERS
         )
+        self.response_transformers = (
+            response_transformers if response_transformers is not None else {}
+        )
 
     def __call__(self, context: AttributePickerContext) -> Any:
         obj = context.obj
@@ -933,6 +966,10 @@ class AliasedAttributePicker(DefaultAttributePicker):
                 break
         else:
             value = None
+        key_path = f"{context.key_path}.{context.key}"
+        if key_path in self.response_transformers:
+            transform = self.response_transformers[key_path]
+            value = transform(value)
         return value
 
     def get_possible_keys(self, context: AttributePickerContext) -> Generator[str]:
@@ -944,7 +981,7 @@ class AliasedAttributePicker(DefaultAttributePicker):
                 yield alias
 
 
-class XFormedAttributePicker(AliasedAttributePicker):
+class XFormedAttributePicker(AttributePicker):
     """Can be injected into a ResponseSerializer to aid in plucking AWS model
     attributes specified in `camelCase` or `PascalCase` from Python objects
     with standard `snake_case` attribute names.
@@ -1089,3 +1126,29 @@ DEFAULT_ALIAS_PROVIDERS = [
     ClassPrefixAlias,
     ModelAlias,
 ]
+
+
+# Response transformers can be used to modify the value of a key in the response.
+def never_return(_: Any) -> None:
+    """
+    A utility function that is used to ensure that certain attributes are never returned
+    in the response. This is useful for attributes that should not be exposed or are not
+    relevant in the context of the response.
+    """
+    return None
+
+
+def return_if_not_empty(value: Any) -> Any:
+    """
+    A utility function that returns the value if it is not empty (i.e., not None, "", {}, or []),
+    otherwise returns None. This is useful for attributes that should only be included in the
+    response if they have a meaningful value.
+    """
+    return value if value not in [None, "", {}, []] else None
+
+
+def url_encode(value: Any) -> Any:
+    """A utility function that url encodes a value before inclusion in a response."""
+    from urllib.parse import quote
+
+    return quote(value) if isinstance(value, str) else value
diff --git a/moto/logs/models.py b/moto/logs/models.py
index 51523c09f..8be2ba4d1 100644
--- a/moto/logs/models.py
+++ b/moto/logs/models.py
@@ -88,7 +88,7 @@ class LogQuery(BaseModel):
     def to_result_json(self) -> Dict[str, Any]:
         return {
             "results": [
-                [{"field": key, "value": val} for key, val in result.items()]
+                [{"field": key, "value": str(val)} for key, val in result.items()]
                 for result in self.results
             ],
             "status": self.status,
diff --git a/moto/rds/exceptions.py b/moto/rds/exceptions.py
index 047741824..fc27a352e 100644
--- a/moto/rds/exceptions.py
+++ b/moto/rds/exceptions.py
@@ -297,3 +297,35 @@ class DBProxyNotFoundFault(RDSClientError):
             "DBProxyNotFoundFault",
             f"The specified proxy name {db_proxy_identifier} doesn't correspond to a proxy owned by your Amazon Web Services account in the specified Amazon Web Services Region.",
         )
+
+
+class BlueGreenDeploymentAlreadyExistsFault(RDSClientError):
+    def __init__(self, bg_name: str):
+        super().__init__(
+            "BlueGreenDeploymentAlreadyExistsFault",
+            f"A blue/green deployment with the specified name {bg_name} already exists.",
+        )
+
+
+class BlueGreenDeploymentNotFoundFault(RDSClientError):
+    def __init__(self, bg_identifier: str):
+        super().__init__(
+            "BlueGreenDeploymentNotFoundFault",
+            f"BlueGreenDeploymentIdentifier {bg_identifier} doesn't refer to an existing blue/green deployment.",
+        )
+
+
+class InvalidBlueGreenDeploymentStateFault(RDSClientError):
+    def __init__(self, bg_identifier: str):
+        super().__init__(
+            "InvalidBlueGreenDeploymentStateFault",
+            f"The blue/green deployment {bg_identifier} can't be switched over or deleted because there is an invalid configuration in the green environment.",
+        )
+
+
+class SourceDatabaseNotSupportedFault(RDSClientError):
+    def __init__(self, source_arn: str):
+        super().__init__(
+            "SourceDatabaseNotSupportedFault",
+            f"The source DB instance {source_arn} isn't supported for a blue/green deployment.",
+        )
diff --git a/moto/rds/models.py b/moto/rds/models.py
index cb4f97375..2f43f8ff2 100644
--- a/moto/rds/models.py
+++ b/moto/rds/models.py
@@ -8,7 +8,9 @@ import re
 import string
 import uuid
 from collections import OrderedDict, defaultdict
+from dataclasses import dataclass
 from datetime import datetime
+from enum import Enum
 from functools import lru_cache, partialmethod
 from re import compile as re_compile
 from typing import (
@@ -36,6 +38,8 @@ from moto.secretsmanager.models import FakeSecret, SecretsManagerBackend
 from moto.utilities.utils import ARN_PARTITION_REGEX, CaseInsensitiveDict, load_resource
 
 from .exceptions import (
+    BlueGreenDeploymentAlreadyExistsFault,
+    BlueGreenDeploymentNotFoundFault,
     DBClusterNotFoundError,
     DBClusterParameterGroupNotFoundError,
     DBClusterSnapshotAlreadyExistsError,
@@ -54,6 +58,7 @@ from .exceptions import (
     DBSubnetGroupNotFoundError,
     ExportTaskAlreadyExistsError,
     ExportTaskNotFoundError,
+    InvalidBlueGreenDeploymentStateFault,
     InvalidDBClusterSnapshotStateFault,
     InvalidDBClusterStateFault,
     InvalidDBClusterStateFaultError,
@@ -72,6 +77,7 @@ from .exceptions import (
     RDSClientError,
     SharedSnapshotQuotaExceeded,
     SnapshotQuotaExceededFault,
+    SourceDatabaseNotSupportedFault,
     SubscriptionAlreadyExistError,
     SubscriptionNotFoundError,
 )
@@ -504,6 +510,9 @@ class DBCluster(RDSBaseModel):
     ):
         super().__init__(backend)
         self.database_name = database_name
+        self.url_identifier = "".join(
+            random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
+        )
         self.db_cluster_identifier = db_cluster_identifier
         self.db_cluster_instance_class = kwargs.get("db_cluster_instance_class")
         self.auto_minor_version_upgrade = auto_minor_version_upgrade
@@ -575,11 +584,6 @@ class DBCluster(RDSBaseModel):
         )
         self.parameter_group = db_cluster_parameter_group_name or default_pg
         self.db_subnet_group_name = db_subnet_group_name or "default"
-        self.url_identifier = "".join(
-            random.choice(string.ascii_lowercase + string.digits) for _ in range(12)
-        )
-        self.endpoint = f"{self.db_cluster_identifier}.cluster-{self.url_identifier}.{self.region}.rds.amazonaws.com"
-        self.reader_endpoint = f"{self.db_cluster_identifier}.cluster-ro-{self.url_identifier}.{self.region}.rds.amazonaws.com"
         self.port = port or DBCluster.default_port(self.engine)
         self.preferred_backup_window = preferred_backup_window
         self.preferred_maintenance_window = preferred_maintenance_window
@@ -787,6 +791,14 @@ class DBCluster(RDSBaseModel):
         self.designate_writer()
         return self._members
 
+    @property
+    def endpoint(self) -> str:
+        return f"{self.db_cluster_identifier}.cluster-{self.url_identifier}.{self.region}.rds.amazonaws.com"
+
+    @property
+    def reader_endpoint(self) -> str:
+        return f"{self.db_cluster_identifier}.cluster-ro-{self.url_identifier}.{self.region}.rds.amazonaws.com"
+
     def designate_writer(self) -> None:
         if self.writer or not self._members:
             return
@@ -1077,6 +1089,7 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
         availability_zone: Optional[str] = None,
         manage_master_user_password: Optional[bool] = False,
         master_user_secret_kms_key_id: Optional[str] = None,
+        storage_throughput: Optional[int] = None,
         **kwargs: Any,
     ) -> None:
         super().__init__(backend)
@@ -1180,6 +1193,7 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
                 option_suffix = option_suffix + "-" + semantic[1]
             default_option_group_name = f"default:{self.engine}-{option_suffix}"
             self.option_group_name = option_group_name or default_option_group_name
+            self.storage_throughput = storage_throughput
 
     @property
     def db_instance_identifier(self) -> str:
@@ -1436,6 +1450,18 @@ class DBInstance(EventMixin, CloudFormationModel, RDSBaseModel):
     def iam_database_authentication_enabled(self) -> bool:
         return self.enable_iam_database_authentication
 
+    @property
+    def storage_throughput(self) -> Optional[int]:
+        return self._storage_throughput
+
+    @storage_throughput.setter
+    def storage_throughput(self, value: Optional[int]) -> None:
+        self._storage_throughput: Optional[int]
+        if value and self.storage_type == "gp3":
+            self._storage_throughput = value
+        else:
+            self._storage_throughput = None
+
     def add_replica(self, replica: DBInstance) -> None:
         if self.region != replica.region:
             # Cross Region replica
@@ -2225,6 +2251,9 @@ class RDSBackend(BaseBackend):
             GlobalCluster: self.global_clusters,
             OptionGroup: self.option_groups,
         }
+        self.blue_green_deployments: MutableMapping[str, BlueGreenDeployment] = (
+            CaseInsensitiveDict()
+        )
 
     @property
     def kms(self) -> KmsBackend:
@@ -3947,6 +3976,106 @@ class RDSBackend(BaseBackend):
         database = self.databases[db_instance_identifier]
         return database.log_file_manager.files
 
+    def create_blue_green_deployment(
+        self,
+        bg_kwargs: Dict[str, Any],
+    ) -> BlueGreenDeployment:
+        bg_name = bg_kwargs.get("blue_green_deployment_name", "")
+
+        for existing_bg in self.blue_green_deployments.values():
+            if existing_bg.blue_green_deployment_name == bg_name:
+                raise BlueGreenDeploymentAlreadyExistsFault(bg_name)
+        bg_deployment: BlueGreenDeployment = BlueGreenDeployment(self, **bg_kwargs)
+        self.blue_green_deployments[bg_deployment.blue_green_deployment_identifier] = (
+            bg_deployment
+        )
+        # update states only for the response of create command
+        bg_copy = copy.copy(bg_deployment)
+        bg_copy.set_status(bg_deployment.SupportedStates.PROVISIONING)
+
+        return bg_copy
+
+    def describe_blue_green_deployments(
+        self,
+        blue_green_deployment_identifier: Optional[str] = None,
+        filters: Any = None,
+    ) -> List[BlueGreenDeployment]:
+        bg_deployments = self.blue_green_deployments
+        if blue_green_deployment_identifier:
+            filters = merge_filters(
+                filters,
+                {
+                    "blue-green-deployment-identifier": [
+                        blue_green_deployment_identifier
+                    ]
+                },
+            )
+        if filters:
+            bg_deployments = self._filter_resources(
+                bg_deployments, filters, BlueGreenDeployment
+            )
+        if blue_green_deployment_identifier and not bg_deployments:
+            raise BlueGreenDeploymentNotFoundFault(blue_green_deployment_identifier)
+        return list(bg_deployments.values())
+
+    def switchover_blue_green_deployment(
+        self,
+        blue_green_deployment_identifier: str,
+        switchover_timeout: Optional[int] = 300,
+    ) -> BlueGreenDeployment:
+        if blue_green_deployment_identifier not in self.blue_green_deployments:
+            raise BlueGreenDeploymentNotFoundFault(blue_green_deployment_identifier)
+
+        bg_deployment = self.blue_green_deployments[blue_green_deployment_identifier]
+
+        if bg_deployment.status != "AVAILABLE":
+            raise InvalidBlueGreenDeploymentStateFault(blue_green_deployment_identifier)
+
+        bg_deployment_before_switch = copy.copy(bg_deployment)
+
+        bg_deployment.switchover()
+        self.blue_green_deployments[blue_green_deployment_identifier] = bg_deployment
+
+        bg_deployment_before_switch.set_status(
+            bg_deployment.SupportedStates.SWITCHOVER_IN_PROGRESS
+        )
+
+        return bg_deployment_before_switch
+
+    def delete_blue_green_deployment(
+        self, blue_green_deployment_identifier: str, delete_target: bool = False
+    ) -> BlueGreenDeployment:
+        if blue_green_deployment_identifier not in self.blue_green_deployments:
+            raise BlueGreenDeploymentNotFoundFault(blue_green_deployment_identifier)
+        bg_deployment = self.blue_green_deployments.pop(
+            blue_green_deployment_identifier
+        )
+
+        if (
+            delete_target
+            and bg_deployment.status
+            != bg_deployment.SupportedStates.SWITCHOVER_COMPLETED.name
+        ):
+            if self._is_cluster(bg_deployment.target):
+                cluster = find_cluster(bg_deployment.target)
+                for member in cluster.members:
+                    self.delete_db_instance(
+                        db_instance_identifier=member.db_instance_identifier
+                    )
+                self.delete_db_cluster(cluster_identifier=cluster.db_cluster_identifier)
+            else:
+                instance = self.find_db_from_id(bg_deployment.target)
+                self.delete_db_instance(
+                    db_instance_identifier=instance.db_instance_identifier
+                )
+
+        bg_deployment.set_status(bg_deployment.SupportedStates.DELETING)
+        bg_deployment.deletion_time = utcnow()
+        return bg_deployment
+
+    def _is_cluster(self, arn: str) -> bool:
+        return arn.split(":")[-2] == "cluster"
+
 
 class OptionGroup(RDSBaseModel):
     resource_type = "og"
@@ -4157,4 +4286,277 @@ class Event:
         self.date = utcnow()
 
 
+class BlueGreenDeployment(RDSBaseModel):
+    class SupportedStates(Enum):
+        PROVISIONING = 1
+        AVAILABLE = 2
+        SWITCHOVER_IN_PROGRESS = 3
+        SWITCHOVER_COMPLETED = 4
+        INVALID_CONFIGURATION = 5
+        SWITCHOVER_FAILED = 6
+        DELETING = 7
+
+    SUPPORTED_FILTERS = {
+        "blue-green-deployment-identifier": FilterDef(
+            ["blue_green_deployment_identifier"],
+            "System-generated BlueGreen Deployment Identifier",
+        ),
+        "blue-green-deployment-name": FilterDef(
+            ["blue_green_deployment_name"],
+            "User-provided Name for BlueGreen Deployment",
+        ),
+        "source": FilterDef(["source"], "Source Database Instance or Cluster"),
+        "target": FilterDef(["target"], "Target Database Instance or Cluster"),
+    }
+
+    def __init__(
+        self,
+        backend: RDSBackend,
+        blue_green_deployment_name: str,
+        source: str,
+        tags: List[Dict[str, str]] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(backend)
+        self.blue_green_deployment_identifier = self._generate_id()
+        self.blue_green_deployment_name = blue_green_deployment_name
+        self.source = source
+        self.target = self._create_green_instance(
+            target_engine_version=kwargs.get("target_engine_version", None),
+            target_db_parameter_group_name=kwargs.get(
+                "target_db_parameter_group_name", None
+            ),
+            target_db_cluster_parameter_group_name=kwargs.get(
+                "target_db_cluster_parameter_group_name", None
+            ),
+            target_db_instance_class=kwargs.get("target_db_instance_class", None),
+            target_iops=kwargs.get("target_iops", None),
+            target_storage_type=kwargs.get("target_storage_type", None),
+            target_allocated_storage=kwargs.get("target_allocated_storage", None),
+            target_storage_throughput=kwargs.get("target_storage_throughput", None),
+        )
+        self.tasks = [
+            BlueGreenDeploymentTask(
+                "CREATING_READ_REPLICA_OF_SOURCE",
+                BlueGreenDeploymentTask.SupportedStates.COMPLETED.name,
+            ),
+            BlueGreenDeploymentTask(
+                "CONFIGURE_BACKUPS",
+                BlueGreenDeploymentTask.SupportedStates.COMPLETED.name,
+            ),
+        ]
+        self.set_status(self.SupportedStates.AVAILABLE)
+        self.status_details: str | None = None
+        self.create_time = utcnow()
+        self.deletion_time: datetime | None = None
+        self.tags = tags or []
+
+    @property
+    def resource_id(self) -> str:
+        return self.blue_green_deployment_identifier
+
+    def set_status(self, value: SupportedStates) -> None:
+        self.status = value.name
+        self.switchover_details = [
+            SwitchoverDetails(self.source, self.target, self.status)
+        ]
+
+    def _generate_id(self) -> str:
+        return "bgd-" + "".join(
+            random.choices(string.ascii_lowercase + string.digits, k=16)
+        )
+
+    def _generate_green_id(self, db_identifier: str) -> str:
+        return f"{db_identifier}-green-" + "".join(
+            random.choices(string.ascii_lowercase, k=6)
+        )
+
+    def _create_green_instance(
+        self,
+        target_engine_version: str | None,
+        target_db_parameter_group_name: str | None,
+        target_db_cluster_parameter_group_name: str | None,
+        target_db_instance_class: str | None,
+        target_iops: int | None,
+        target_storage_type: str | None,
+        target_allocated_storage: int | None,
+        target_storage_throughput: int | None,
+    ) -> str:
+        source_instance: DBInstance | DBCluster = self._get_instance(self.source)
+        green_instance: DBInstance | DBCluster
+        if source_instance.manage_master_user_password:
+            raise SourceDatabaseNotSupportedFault(source_instance.arn)
+
+        db_kwargs = {
+            "engine": source_instance.engine,
+            "engine_version": target_engine_version or source_instance.engine_version,
+            "iops": target_iops or source_instance.iops,
+            "port": source_instance.port,
+            "backup_retention_period": source_instance.backup_retention_period,
+            "character_set_name": source_instance.character_set_name,
+            "auto_minor_version_upgrade": source_instance.auto_minor_version_upgrade,
+            "copy_tags_to_snapshot": source_instance.copy_tags_to_snapshot,
+            "master_username": source_instance.master_username,
+            "multi_az": source_instance.multi_az,
+            "license_model": source_instance.license_model,
+            "preferred_backup_window": source_instance.preferred_backup_window,
+            "preferred_maintenance_window": source_instance.preferred_maintenance_window,
+            "storage_encrypted": source_instance.storage_encrypted,
+            "tags": source_instance.tags,
+            "deletion_protection": source_instance.deletion_protection,
+            "enable_cloudwatch_logs_exports": source_instance.enabled_cloudwatch_logs_exports,
+            "master_user_password": source_instance._master_user_password,
+            "kms_key_id": source_instance.kms_key_id,
+        }
+
+        if isinstance(source_instance, DBInstance):
+            db_instance_specific_kwargs = {
+                "storage_type": target_storage_type or source_instance.storage_type,
+                "allocated_storage": target_allocated_storage
+                or source_instance.allocated_storage,
+                "db_instance_identifier": self._generate_green_id(
+                    source_instance.db_instance_identifier
+                ),
+                "db_instance_class": target_db_instance_class
+                or source_instance.db_instance_class,
+                "storage_throughput": target_storage_throughput
+                or source_instance.storage_throughput
+                or None,
+                "max_allocated_storage": source_instance.max_allocated_storage,
+                "db_security_groups": source_instance.db_security_groups,
+                "db_cluster_identifier": source_instance.db_cluster_identifier,
+                "publicly_accessible": source_instance.publicly_accessible,
+                "source_db_instance_identifier": source_instance.source_db_instance_identifier,
+                "vpc_security_group_ids": source_instance.vpc_security_group_ids,
+                "ca_certificate_identifier": source_instance.ca_certificate_identifier,
+                "availability_zone": source_instance.availability_zone,
+                "db_subnet_group_name": source_instance._db_subnet_group_name,
+                "db_name": source_instance.db_name,
+            }
+            if (
+                target_db_parameter_group_name
+                or source_instance.db_parameter_group_name
+            ):
+                db_instance_specific_kwargs["db_parameter_group_name"] = (
+                    target_db_parameter_group_name
+                    or source_instance.db_parameter_group_name
+                )
+
+            green_instance = self.backend.create_db_instance(
+                db_kwargs | db_instance_specific_kwargs
+            )
+            return green_instance.db_instance_arn
+        else:
+            db_cluster_specific_kwargs = {
+                "db_cluster_identifier": self._generate_green_id(
+                    source_instance.db_cluster_identifier
+                ),
+                "allocated_storage": source_instance.allocated_storage,
+                "database_name": source_instance.database_name,
+                "db_cluster_parameter_group_name": target_db_cluster_parameter_group_name
+                or source_instance.db_cluster_parameter_group,
+                "vpc_security_group_ids": source_instance._vpc_security_group_ids,
+            }
+
+            green_instance = self.backend.create_db_cluster(
+                db_kwargs | db_cluster_specific_kwargs
+            )
+
+            for cluster_member in source_instance.members:
+                cluster_member_specific_kwargs = {
+                    "storage_type": cluster_member.storage_type,
+                    "allocated_storage": cluster_member.allocated_storage,
+                    "db_instance_identifier": self._generate_green_id(
+                        cluster_member.db_instance_identifier
+                    ),
+                    "db_cluster_identifier": green_instance.db_cluster_identifier,
+                    "db_instance_class": cluster_member.db_instance_class,
+                    "max_allocated_storage": cluster_member.max_allocated_storage,
+                    "publicly_accessible": cluster_member.publicly_accessible,
+                    "availability_zone": cluster_member.availability_zone,
+                    "db_subnet_group_name": cluster_member._db_subnet_group_name,
+                }
+                if (
+                    target_db_parameter_group_name
+                    or cluster_member.db_parameter_group_name
+                ):
+                    cluster_member_specific_kwargs["db_parameter_group_name"] = (
+                        target_db_parameter_group_name
+                        or cluster_member.db_parameter_group_name
+                    )
+
+                self.backend.create_db_instance(
+                    db_kwargs | cluster_member_specific_kwargs
+                )
+            return green_instance.db_cluster_arn
+
+    def _get_instance(self, arn: str) -> DBInstance | DBCluster:
+        result = re.findall(r"^arn:[A-Za-z][0-9A-Za-z-:._]*", arn)
+        if len(result) == 0:
+            raise InvalidParameterValue("Provided string is not a valid arn")
+        if self.backend._is_cluster(arn):
+            return find_cluster(arn)
+        else:
+            return self.backend.find_db_from_id(arn)
+
+    def switchover(self) -> None:
+        source = self._get_instance(self.source)
+        target = self._get_instance(self.target)
+        source_result: DBCluster | DBInstance
+        target_result: DBCluster | DBInstance
+
+        if isinstance(source, DBCluster):
+            new_target_name = source.db_cluster_identifier
+            source_result = self.backend.modify_db_cluster(
+                {
+                    "db_cluster_identifier": source.db_cluster_identifier,
+                    "new_db_cluster_identifier": f"{source.db_cluster_identifier}-old1",
+                }
+            )
+        else:
+            new_target_name = source.db_instance_identifier
+            source_result = self.backend.modify_db_instance(
+                db_instance_identifier=source.db_instance_identifier,
+                db_kwargs={
+                    "new_db_instance_identifier": f"{source.db_instance_identifier}-old1"
+                },
+            )
+
+        if isinstance(target, DBCluster):
+            target_result = self.backend.modify_db_cluster(
+                {
+                    "db_cluster_identifier": target.db_cluster_identifier,
+                    "new_db_cluster_identifier": new_target_name,
+                }
+            )
+        else:
+            target_result = self.backend.modify_db_instance(
+                db_instance_identifier=target.db_instance_identifier,
+                db_kwargs={"new_db_instance_identifier": new_target_name},
+            )
+        self.source = source_result.arn
+        self.target = target_result.arn
+        self.set_status(self.SupportedStates.SWITCHOVER_COMPLETED)
+        self.status_details = "Switchover completed"
+
+
+@dataclass
+class SwitchoverDetails:
+    SourceMember: str
+    TargetMember: str
+    Status: str
+
+
+@dataclass
+class BlueGreenDeploymentTask:
+    Name: str
+    Status: str
+
+    class SupportedStates(Enum):
+        PENDING = 1
+        IN_PROGRESS = 2
+        COMPLETED = 3
+        FAILED = 4
+
+
 rds_backends = BackendDict(RDSBackend, "rds")
diff --git a/moto/rds/responses.py b/moto/rds/responses.py
index c568be37a..af70c00a0 100644
--- a/moto/rds/responses.py
+++ b/moto/rds/responses.py
@@ -781,6 +781,33 @@ class RDSResponse(BaseResponse):
         result = {"DescribeDBLogFiles": log_files}
         return ActionResult(result)
 
+    def create_blue_green_deployment(self) -> ActionResult:
+        bg_kwargs = self.parameters
+        bg_deployment = self.backend.create_blue_green_deployment(bg_kwargs)
+        result = {"BlueGreenDeployment": bg_deployment}
+        return ActionResult(result)
+
+    def describe_blue_green_deployments(self) -> ActionResult:
+        filters = self.parameters.get("Filters", [])
+        filter_dict = {f["Name"]: f["Values"] for f in filters}
+        self.parameters["filters"] = filter_dict
+        all_bg_deployments = self.backend.describe_blue_green_deployments(
+            **self.parameters
+        )
+        bg_deployments, _ = self._paginate(all_bg_deployments)
+        result = {"BlueGreenDeployments": bg_deployments}
+        return ActionResult(result)
+
+    def switchover_blue_green_deployment(self) -> ActionResult:
+        bg_deployment = self.backend.switchover_blue_green_deployment(**self.parameters)
+        result = {"BlueGreenDeployment": bg_deployment}
+        return ActionResult(result)
+
+    def delete_blue_green_deployment(self) -> ActionResult:
+        bg_deployment = self.backend.delete_blue_green_deployment(**self.parameters)
+        result = {"BlueGreenDeployment": bg_deployment}
+        return ActionResult(result)
+
     def _paginate(self, resources: List[Any]) -> Tuple[List[Any], Optional[str]]:
         from moto.rds.exceptions import InvalidParameterValue
 
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 3ea17db67..ecfdd27d0 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -229,6 +229,14 @@ class FakeKey(BaseModel, ManagedState):
     def set_metadata(self, metadata: Any, replace: bool = False) -> None:
         if replace:
             self._metadata = {}  # type: ignore
+        # Remove AWS-specific Content-Encoding
+        if encoding := metadata.pop("Content-Encoding", None):
+            # Remove 'aws-chunked', but keep any other (user-provided) content encoding
+            encoding = ",".join(
+                [enc for enc in encoding.split(",") if enc != "aws-chunked"]
+            )
+            if encoding:
+                metadata["Content-Encoding"] = encoding
         self._metadata.update(metadata)
 
     def set_storage_class(self, storage: Optional[str]) -> None:
diff --git a/moto/secretsmanager/list_secrets/filters.py b/moto/secretsmanager/list_secrets/filters.py
index 87f05613e..da35e4864 100644
--- a/moto/secretsmanager/list_secrets/filters.py
+++ b/moto/secretsmanager/list_secrets/filters.py
@@ -2,7 +2,7 @@ import re
 from typing import TYPE_CHECKING, Iterator, List, Union
 
 if TYPE_CHECKING:
-    from ..models import FakeSecret
+    from ..models import FakeSecret, ReplicaSecret
 
 
 def name_filter(secret: "FakeSecret", names: List[str]) -> bool:
@@ -29,19 +29,21 @@ def owning_service_filter(secret: "FakeSecret", owning_services: List[str]) -> b
     )
 
 
-def tag_key(secret: "FakeSecret", tag_keys: List[str]) -> bool:
+def tag_key(secret: Union["FakeSecret", "ReplicaSecret"], tag_keys: List[str]) -> bool:
     if not secret.tags:
         return False
     return _matcher(tag_keys, [tag["Key"] for tag in secret.tags])
 
 
-def tag_value(secret: "FakeSecret", tag_values: List[str]) -> bool:
+def tag_value(
+    secret: Union["FakeSecret", "ReplicaSecret"], tag_values: List[str]
+) -> bool:
     if not secret.tags:
         return False
     return _matcher(tag_values, [tag["Value"] for tag in secret.tags])
 
 
-def filter_all(secret: "FakeSecret", values: List[str]) -> bool:
+def filter_all(secret: Union["FakeSecret", "ReplicaSecret"], values: List[str]) -> bool:
     attributes = [secret.name]
     if secret.description:
         attributes.append(secret.description)
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index 36ee0c5ac..dff10f48d 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -36,7 +36,9 @@ from .utils import (
 MAX_RESULTS_DEFAULT = 100
 
 
-def filter_primary_region(secret: "FakeSecret", values: List[str]) -> bool:
+def filter_primary_region(
+    secret: Union["FakeSecret", "ReplicaSecret"], values: List[str]
+) -> bool:
     if isinstance(secret, FakeSecret):
         return len(secret.replicas) > 0 and secret.region in values
     elif isinstance(secret, ReplicaSecret):
@@ -330,6 +332,18 @@ class ReplicaSecret:
     def secret_id(self) -> str:
         return self.source.secret_id
 
+    @property
+    def description(self) -> Optional[str]:
+        return self.source.description
+
+    @property
+    def tags(self) -> Optional[List[Dict[str, str]]]:
+        return self.source.tags
+
+    @property
+    def owning_service(self) -> Optional[str]:
+        return self.source.owning_service
+
 
 class SecretsStore(Dict[str, Union[FakeSecret, ReplicaSecret]]):
     # Parameters to this dictionary can be three possible values:


Test patch:
diff --git a/tests/test_core/test_response_serialization.py b/tests/test_core/test_response_serialization.py
new file mode 100644
index 000000000..7d68bc9c7
--- /dev/null
+++ b/tests/test_core/test_response_serialization.py
@@ -0,0 +1,60 @@
+from botocore.model import ServiceModel
+
+from moto.core.responses import ActionContext, ActionResult, BaseResponse, EmptyResult
+from moto.core.serialize import QuerySerializer, never_return
+
+TEST_MODEL = {
+    "metadata": {"protocol": "query", "apiVersion": "2014-01-01"},
+    "documentation": "",
+    "operations": {
+        "TestOperation": {
+            "name": "TestOperation",
+            "http": {
+                "method": "POST",
+                "requestUri": "/",
+            },
+            "output": {"shape": "OutputShape"},
+        }
+    },
+    "shapes": {
+        "OutputShape": {
+            "type": "structure",
+            "members": {
+                "string": {"shape": "StringType"},
+                "TransformerTest": {"shape": "StringType"},
+            },
+        },
+        "StringType": {
+            "type": "string",
+        },
+    },
+}
+
+
+def test_response_result_execution() -> None:
+    """Exercise various response result execution paths."""
+
+    class TestResponseClass(BaseResponse):
+        RESPONSE_KEY_PATH_TO_TRANSFORMER = {"OutputShape.TransformerTest": never_return}
+
+    class TestResponseObject:
+        string = "test-string"
+        TransformerTest = "some data"
+
+    service_model = ServiceModel(TEST_MODEL)
+    operation_model = service_model.operation_model("TestOperation")
+    serializer_class = QuerySerializer
+    response_class = TestResponseClass
+
+    context = ActionContext(
+        service_model, operation_model, serializer_class, response_class
+    )
+
+    result = ActionResult(TestResponseObject())
+    _, __, body = result.execute_result(context)
+    assert "test-string" in body
+    assert "TransformerTest" not in body
+
+    result = EmptyResult()
+    _, __, body = result.execute_result(context)
+    assert "<OutputShapeResult/>" in body
diff --git a/tests/test_core/test_serialize.py b/tests/test_core/test_serialize.py
index 96fda31cb..a6e86a0f9 100644
--- a/tests/test_core/test_serialize.py
+++ b/tests/test_core/test_serialize.py
@@ -21,6 +21,9 @@ from moto.core.serialize import (
     AttributePickerContext,
     QuerySerializer,
     XFormedAttributePicker,
+    never_return,
+    return_if_not_empty,
+    url_encode,
 )
 
 
@@ -217,3 +220,25 @@ class TestXFormedAttributePicker:
         ctx = AttributePickerContext(DBInstance(), "DBInstanceClass", None)
         value = self.picker(ctx)
         assert value == "t2.medium"
+
+
+class TestResponseAttributeTransformers:
+    def test_never_return(self):
+        assert never_return(None) is None
+        assert never_return(True) is None
+        assert never_return(False) is None
+        assert never_return("Test") is None
+
+    def test_return_if_not_empty(self):
+        assert return_if_not_empty([]) is None
+        assert return_if_not_empty(None) is None
+        assert return_if_not_empty("Test") == "Test"
+        assert return_if_not_empty({}) is None
+        assert return_if_not_empty("") is None
+        assert return_if_not_empty({"key": "value"}) == {"key": "value"}
+        assert return_if_not_empty([0, 1, 2]) == [0, 1, 2]
+
+    def test_url_encode(self):
+        assert url_encode(None) is None
+        assert url_encode("Test") == "Test"
+        assert url_encode("Test String") == "Test%20String"
diff --git a/tests/test_logs/test_logs_query/test_boto3.py b/tests/test_logs/test_logs_query/test_boto3.py
index 85ca61a63..4bff5a50c 100644
--- a/tests/test_logs/test_logs_query/test_boto3.py
+++ b/tests/test_logs/test_logs_query/test_boto3.py
@@ -61,7 +61,7 @@ def test_get_query_results():
     )
 
     query_id = client.start_query(
-        logGroupName="test",
+        logGroupName=log_group_name,
         startTime=int(unix_time(utcnow() - timedelta(minutes=10))),
         endTime=int(unix_time(utcnow() + timedelta(minutes=10))),
         queryString="fields @message",
@@ -90,7 +90,7 @@ def test_get_query_results():
 
     # Only find events from last 2 minutes
     query_id = client.start_query(
-        logGroupName="test",
+        logGroupName=log_group_name,
         startTime=int(unix_time(utcnow() - timedelta(minutes=2, seconds=1))),
         endTime=int(unix_time(utcnow() - timedelta(seconds=1))),
         queryString="fields @message",
@@ -98,6 +98,7 @@ def test_get_query_results():
 
     resp = client.get_query_results(queryId=query_id)
     assert len(resp["results"]) == 2
+    assert isinstance(resp["results"][0][0]["value"], str)
 
     messages = [
         row["value"]
diff --git a/tests/test_rds/test_rds_blue_green_deployment.py b/tests/test_rds/test_rds_blue_green_deployment.py
new file mode 100644
index 000000000..86fdf49fc
--- /dev/null
+++ b/tests/test_rds/test_rds_blue_green_deployment.py
@@ -0,0 +1,762 @@
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from moto import mock_aws
+
+from . import DEFAULT_REGION
+from .test_rds import (
+    create_db_instance,
+    create_subnet,
+)
+from .test_rds_clusters import create_db_cluster
+
+
+@pytest.fixture(name="client")
+@mock_aws
+def get_rds_client():
+    return boto3.client("rds", region_name=DEFAULT_REGION)
+
+
+@mock_aws
+def test_create_bluegreen_deployment_creates_a_green_db_instance(client):
+    expected_target_engine_version = "17.3"
+    expected_target_db_instance_class = "db.6g.xlarge"
+    expected_target_allocated_storage = 999
+    expected_db_parameter_group_name = "new_parameter_group"
+    expected_target_iops = 5000
+    expected_target_storage_type = "gp3"
+    expected_target_storage_throughput = 6329
+
+    subnet_id = create_subnet()
+    subnet = client.create_db_subnet_group(
+        DBSubnetGroupName="db_subnet1",
+        DBSubnetGroupDescription="my db subnet",
+        SubnetIds=[subnet_id],
+        Tags=[{"Value": "bar", "Key": "foo"}, {"Value": "bar1", "Key": "foo1"}],
+    )["DBSubnetGroup"]["DBSubnetGroupName"]
+
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+        DBSubnetGroupName=subnet,
+        MasterUsername="Bob",
+        ManageMasterUserPassword=False,
+    )
+    client.create_db_parameter_group(
+        DBParameterGroupName=expected_db_parameter_group_name,
+        DBParameterGroupFamily="postgres",
+        Description="Foobar",
+    )
+
+    response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName="FooBarBlueGreen",
+        Source=instance["DBInstanceArn"],
+        TargetEngineVersion=expected_target_engine_version,
+        TargetDBInstanceClass=expected_target_db_instance_class,
+        TargetAllocatedStorage=expected_target_allocated_storage,
+        TargetDBParameterGroupName=expected_db_parameter_group_name,
+        TargetIops=expected_target_iops,
+        TargetStorageType=expected_target_storage_type,
+        TargetStorageThroughput=expected_target_storage_throughput,
+    )
+
+    bluegreen_deployment_response = response["BlueGreenDeployment"]
+
+    db_describe_response = client.describe_db_instances(
+        DBInstanceIdentifier=bluegreen_deployment_response["Target"]
+    )
+
+    assert bluegreen_deployment_response is not None
+    assert "bgd-" in bluegreen_deployment_response["BlueGreenDeploymentIdentifier"]
+    assert bluegreen_deployment_response["BlueGreenDeploymentName"] == "FooBarBlueGreen"
+    assert bluegreen_deployment_response["Source"] == instance["DBInstanceArn"]
+    assert (
+        bluegreen_deployment_response["Target"]
+        == db_describe_response["DBInstances"][0]["DBInstanceArn"]
+    )
+    assert bluegreen_deployment_response["Status"] == "PROVISIONING"
+
+    assert len(db_describe_response["DBInstances"]) > 0
+    assert (
+        db_describe_response["DBInstances"][0]["EngineVersion"]
+        == expected_target_engine_version
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["DBInstanceClass"]
+        == expected_target_db_instance_class
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["AllocatedStorage"]
+        == expected_target_allocated_storage
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["DBParameterGroups"][0][
+            "DBParameterGroupName"
+        ]
+        == expected_db_parameter_group_name
+    )
+    assert db_describe_response["DBInstances"][0]["Iops"] == expected_target_iops
+    assert (
+        db_describe_response["DBInstances"][0]["StorageType"]
+        == expected_target_storage_type
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["StorageThroughput"]
+        == expected_target_storage_throughput
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["DBSecurityGroups"]
+        == instance["DBSecurityGroups"]
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["DBSubnetGroup"]["DBSubnetGroupName"]
+        == instance["DBSubnetGroup"]["DBSubnetGroupName"]
+    )
+    assert (
+        db_describe_response["DBInstances"][0]["OptionGroupMemberships"][0][
+            "OptionGroupName"
+        ]
+        != instance["OptionGroupMemberships"][0]["OptionGroupName"]
+    )
+    assert db_describe_response["DBInstances"][0].get("MasterUserSecret") is None
+
+
+@mock_aws
+def test_create_bluegreen_deployment_creates_a_green_db_cluster(client):
+    expected_target_engine_version = "17.3"
+    expected_db_parameter_group_name = "new_parameter_group"
+    expected_db_cluster_parameter_group_name = "new_cluster_parameter_group"
+
+    subnet_id = create_subnet()
+    subnet = client.create_db_subnet_group(
+        DBSubnetGroupName="db_subnet1",
+        DBSubnetGroupDescription="my db subnet",
+        SubnetIds=[subnet_id],
+        Tags=[{"Value": "bar", "Key": "foo"}, {"Value": "bar1", "Key": "foo1"}],
+    )["DBSubnetGroup"]["DBSubnetGroupName"]
+
+    create_db_cluster(
+        DBClusterIdentifier="cluster-1",
+        DBSubnetGroupName=subnet,
+        MasterUsername="Bob",
+        Engine="aurora-postgresql",
+    )
+    clustered_instances = []
+    for i in range(3):
+        clustered_instances.append(
+            client.create_db_instance(
+                DBInstanceIdentifier=f"test-instance-{i}",
+                DBInstanceClass="db.m1.small",
+                Engine="aurora-postgresql",
+                DBClusterIdentifier="cluster-1",
+                DBSubnetGroupName=subnet,
+            )
+        )
+    cluster = client.describe_db_clusters(
+        DBClusterIdentifier="cluster-1",
+    )["DBClusters"][0]
+    client.create_db_parameter_group(
+        DBParameterGroupName=expected_db_parameter_group_name,
+        DBParameterGroupFamily="postgres",
+        Description="Foobar",
+    )
+    client.create_db_cluster_parameter_group(
+        DBClusterParameterGroupName=expected_db_cluster_parameter_group_name,
+        DBParameterGroupFamily="aurora-postgresql14",
+        Description="FooBar",
+    )
+
+    response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName="FooBarBlueGreen",
+        Source=cluster["DBClusterArn"],
+        TargetEngineVersion=expected_target_engine_version,
+        TargetDBParameterGroupName=expected_db_parameter_group_name,
+        TargetDBClusterParameterGroupName=expected_db_cluster_parameter_group_name,
+    )
+
+    bluegreen_deployment_response = response["BlueGreenDeployment"]
+
+    db_describe_response = client.describe_db_clusters(
+        DBClusterIdentifier=bluegreen_deployment_response["Target"]
+    )
+
+    assert bluegreen_deployment_response is not None
+    assert "bgd-" in bluegreen_deployment_response["BlueGreenDeploymentIdentifier"]
+    assert bluegreen_deployment_response["BlueGreenDeploymentName"] == "FooBarBlueGreen"
+    assert bluegreen_deployment_response["Source"] == cluster["DBClusterArn"]
+    assert (
+        bluegreen_deployment_response["Target"]
+        == db_describe_response["DBClusters"][0]["DBClusterArn"]
+    )
+    assert bluegreen_deployment_response["Status"] == "PROVISIONING"
+
+    assert len(db_describe_response["DBClusters"]) > 0
+    assert (
+        db_describe_response["DBClusters"][0]["EngineVersion"]
+        == expected_target_engine_version
+    )
+    assert (
+        db_describe_response["DBClusters"][0]["AllocatedStorage"]
+        == cluster["AllocatedStorage"]
+    )
+    assert db_describe_response["DBClusters"][0].get("MasterUserSecret") is None
+    assert (
+        db_describe_response["DBClusters"][0]["DBClusterParameterGroup"]
+        == expected_db_cluster_parameter_group_name
+    )
+
+    assert len(db_describe_response["DBClusters"][0]["DBClusterMembers"]) == len(
+        cluster["DBClusterMembers"]
+    )
+    for index, member in enumerate(
+        db_describe_response["DBClusters"][0]["DBClusterMembers"]
+    ):
+        member_response = client.describe_db_instances(
+            DBInstanceIdentifier=member["DBInstanceIdentifier"]
+        )
+
+        assert (
+            member_response["DBInstances"][0]["DBSubnetGroup"]["DBSubnetGroupName"]
+            == clustered_instances[index]["DBInstance"]["DBSubnetGroup"][
+                "DBSubnetGroupName"
+            ]
+        )
+
+
+@mock_aws
+def test_create_blue_green_deployment_db_instance_with_managed_master_password(client):
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+        MasterUsername="Bob",
+        ManageMasterUserPassword=True,
+    )
+
+    source_arn = instance["DBInstanceArn"]
+
+    with pytest.raises(ClientError) as exc:
+        client.create_blue_green_deployment(
+            BlueGreenDeploymentName="FooBarBlueGreen",
+            Source=source_arn,
+        )
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "SourceDatabaseNotSupportedFault"
+    assert (
+        err["Message"]
+        == f"The source DB instance {source_arn} isn't supported for a blue/green deployment."
+    )
+
+
+@mock_aws
+def test_create_blue_green_deployment_rds_cluster_with_managed_master_password(client):
+    create_db_cluster(
+        DBClusterIdentifier="cluster-1",
+        MasterUsername="Bob",
+        ManageMasterUserPassword=True,
+        Engine="aurora-postgresql",
+    )
+    clustered_instances = []
+    for i in range(3):
+        clustered_instances.append(
+            client.create_db_instance(
+                DBInstanceIdentifier=f"test-instance-{i}",
+                DBInstanceClass="db.m1.small",
+                Engine="aurora-postgresql",
+                DBClusterIdentifier="cluster-1",
+            )
+        )
+    cluster = client.describe_db_clusters(
+        DBClusterIdentifier="cluster-1",
+    )["DBClusters"][0]
+
+    source_arn = cluster["DBClusterArn"]
+
+    with pytest.raises(ClientError) as exc:
+        client.create_blue_green_deployment(
+            BlueGreenDeploymentName="FooBarBlueGreen",
+            Source=source_arn,
+        )
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "SourceDatabaseNotSupportedFault"
+    assert (
+        err["Message"]
+        == f"The source DB instance {source_arn} isn't supported for a blue/green deployment."
+    )
+
+
+@mock_aws
+def test_create_blue_green_deployment_error_if_source_db_instance_not_found(client):
+    with pytest.raises(ClientError) as ex:
+        client.create_blue_green_deployment(
+            BlueGreenDeploymentName="FooBarBlueGreen",
+            Source="arn:rds:123456789012:us-east-1:db:not_an_arn",
+        )
+
+    err = ex.value.response["Error"]
+    assert err["Code"] == "DBInstanceNotFound"
+    assert (
+        err["Message"]
+        == "DBInstance arn:rds:123456789012:us-east-1:db:not_an_arn not found."
+    )
+
+
+@mock_aws
+def test_create_blue_green_deployment_error_if_source_rds_cluster_not_found(client):
+    with pytest.raises(ClientError) as ex:
+        client.create_blue_green_deployment(
+            BlueGreenDeploymentName="FooBarBlueGreen",
+            Source="arn:rds:123456789012:us-east-1:cluster:not_an_arn",
+        )
+
+    err = ex.value.response["Error"]
+    assert err["Code"] == "DBClusterNotFoundFault"
+    assert (
+        err["Message"]
+        == "DBCluster arn:rds:123456789012:us-east-1:cluster:not_an_arn not found."
+    )
+
+
+@mock_aws
+def test_create_blue_green_deployment_with_duplicate_name(client):
+    bluegreen_name = "FooBarBlueGreen"
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+    )
+
+    response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bluegreen_name,
+        Source=instance["DBInstanceArn"],
+    )
+
+    with pytest.raises(ClientError) as ex:
+        client.create_blue_green_deployment(
+            BlueGreenDeploymentName=bluegreen_name,
+            Source=instance["DBInstanceArn"],
+        )
+
+    error = ex.value.response["Error"]
+
+    assert response["BlueGreenDeployment"]["BlueGreenDeploymentName"] == bluegreen_name
+    assert error["Code"] == "BlueGreenDeploymentAlreadyExistsFault"
+    assert (
+        error["Message"]
+        == "A blue/green deployment with the specified name FooBarBlueGreen already exists."
+    )
+
+
+@mock_aws
+def test_describe_bluegreen_deployments_db_instance_with_filters(client):
+    bg_name = "bluegreen-deployment-1"
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+    )
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=instance["DBInstanceArn"],
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_with_filters_response = client.describe_blue_green_deployments(
+        Filters=[
+            {"Name": "blue-green-deployment-name", "Values": [bg_name]},
+            {"Name": "source", "Values": [instance["DBInstanceArn"]]},
+        ]
+    )
+
+    update_status_from_create_blue_green_response(
+        create_response["BlueGreenDeployment"]
+    )
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Tasks"][0]["Status"]
+        == "COMPLETED"
+    )
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]
+        == create_response["BlueGreenDeployment"]
+    )
+    assert len(describe_with_filters_response["BlueGreenDeployments"]) > 0
+    assert (
+        describe_with_filters_response["BlueGreenDeployments"][0]
+        == describe_response["BlueGreenDeployments"][0]
+    )
+
+
+@mock_aws
+def test_describe_bluegreen_deployments_db_cluster_with_filters(client):
+    bg_name = "bluegreen-deployment-1"
+    create_db_cluster(
+        DBClusterIdentifier="cluster-1",
+    )
+
+    cluster = client.describe_db_clusters(
+        DBClusterIdentifier="cluster-1",
+    )["DBClusters"][0]
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=cluster["DBClusterArn"],
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_with_filters_response = client.describe_blue_green_deployments(
+        Filters=[
+            {"Name": "blue-green-deployment-name", "Values": [bg_name]},
+            {"Name": "source", "Values": [cluster["DBClusterArn"]]},
+        ]
+    )
+
+    update_status_from_create_blue_green_response(
+        create_response["BlueGreenDeployment"]
+    )
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Tasks"][0]["Status"]
+        == "COMPLETED"
+    )
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]
+        == create_response["BlueGreenDeployment"]
+    )
+    assert len(describe_with_filters_response["BlueGreenDeployments"]) > 0
+    assert (
+        describe_with_filters_response["BlueGreenDeployments"][0]
+        == create_response["BlueGreenDeployment"]
+    )
+
+
+@mock_aws
+def test_describe_bluegreen_deployments_with_no_hits(client):
+    with pytest.raises(ClientError) as ex:
+        client.describe_blue_green_deployments(
+            BlueGreenDeploymentIdentifier="red-yellow"
+        )
+
+    filter_response = client.describe_blue_green_deployments(
+        Filters=[{"Name": "source", "Values": ["no-real-arn"]}]
+    )
+
+    error = ex.value.response["Error"]
+
+    assert error["Code"] == "BlueGreenDeploymentNotFoundFault"
+    assert len(filter_response["BlueGreenDeployments"]) == 0
+
+
+@mock_aws
+def test_switchover_blue_green_deployment_switches_db_instances(client):
+    bg_name = "bluegreen-deployment-1"
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+    )
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=instance["DBInstanceArn"],
+    )
+
+    switchover_response = client.switchover_blue_green_deployment(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_target_instance = client.describe_db_instances(
+        DBInstanceIdentifier=describe_response["BlueGreenDeployments"][0]["Target"]
+    )
+
+    assert (
+        switchover_response["BlueGreenDeployment"]["Status"] == "SWITCHOVER_IN_PROGRESS"
+    )
+    assert (
+        switchover_response["BlueGreenDeployment"]["Source"]
+        == instance["DBInstanceArn"]
+    )
+    assert "green" in switchover_response["BlueGreenDeployment"]["Target"]
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Status"] == "SWITCHOVER_COMPLETED"
+    )
+    assert "old" in describe_response["BlueGreenDeployments"][0]["Source"]
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Target"].split(":")[-1]
+        == instance["DBInstanceIdentifier"]
+    )
+    assert (
+        describe_target_instance["DBInstances"][0]["Endpoint"] == instance["Endpoint"]
+    )
+
+
+@mock_aws
+def test_switchover_blue_green_deployment_switches_rds_clusters(client):
+    bg_name = "bluegreen-deployment-1"
+    create_db_cluster(
+        DBClusterIdentifier="cluster-1",
+    )
+    cluster = client.describe_db_clusters(
+        DBClusterIdentifier="cluster-1",
+    )["DBClusters"][0]
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=cluster["DBClusterArn"],
+    )
+
+    switchover_response = client.switchover_blue_green_deployment(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    describe_target_instance = client.describe_db_clusters(
+        DBClusterIdentifier=describe_response["BlueGreenDeployments"][0]["Target"]
+    )
+
+    assert (
+        switchover_response["BlueGreenDeployment"]["Status"] == "SWITCHOVER_IN_PROGRESS"
+    )
+    assert (
+        switchover_response["BlueGreenDeployment"]["Source"] == cluster["DBClusterArn"]
+    )
+    assert "green" in switchover_response["BlueGreenDeployment"]["Target"]
+
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Status"] == "SWITCHOVER_COMPLETED"
+    )
+    assert "old" in describe_response["BlueGreenDeployments"][0]["Source"]
+    assert (
+        describe_response["BlueGreenDeployments"][0]["Target"].split(":")[-1]
+        == cluster["DBClusterIdentifier"]
+    )
+    assert (
+        describe_target_instance["DBClusters"][0]["Endpoint"].split(".")[0]
+        == cluster["Endpoint"].split(".")[0]
+    )
+
+
+@mock_aws
+def test_switchover_after_successful_switchover_blue_green_deployment(client):
+    bg_name = "bluegreen-deployment-1"
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+    )
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=instance["DBInstanceArn"],
+    )
+
+    client.switchover_blue_green_deployment(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ]
+    )
+
+    with pytest.raises(ClientError) as exc:
+        client.switchover_blue_green_deployment(
+            BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+                "BlueGreenDeploymentIdentifier"
+            ]
+        )
+
+    error = exc.value.response["Error"]
+    assert error["Code"] == "InvalidBlueGreenDeploymentStateFault"
+
+
+@mock_aws
+@pytest.mark.parametrize(
+    "options",
+    [
+        pytest.param(
+            {"with_switchover": False, "delete_target": True},
+            id="before_switchover_with_delete_target",
+        ),
+        pytest.param(
+            {"with_switchover": False, "delete_target": False},
+            id="before_switchover_without_delete_target",
+        ),
+        pytest.param(
+            {"with_switchover": True, "delete_target": True},
+            id="after_switchover_with_delete_target",
+        ),
+    ],
+)
+def test_delete_blue_green_deployment_with_source_db_instance(client, options):
+    bg_name = "bluegreen-deployment-1"
+
+    instance = create_db_instance(
+        DBInstanceIdentifier="FooBar",
+    )
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=instance["DBInstanceArn"],
+    )
+
+    if options["with_switchover"]:
+        client.switchover_blue_green_deployment(
+            BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+                "BlueGreenDeploymentIdentifier"
+            ]
+        )
+
+    delete_response = client.delete_blue_green_deployment(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ],
+        DeleteTarget=options["delete_target"],
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        Filters=[{"Name": "blue-green-deployment-name", "Values": [bg_name]}]
+    )
+
+    describe_source_instance = client.describe_db_instances(
+        DBInstanceIdentifier=delete_response["BlueGreenDeployment"]["Source"]
+    )
+
+    describe_target_instance = client.describe_db_instances(
+        Filters=[
+            {
+                "Name": "db-instance-id",
+                "Values": [delete_response["BlueGreenDeployment"]["Target"]],
+            }
+        ]
+    )
+
+    assert delete_response["BlueGreenDeployment"]["Status"] == "DELETING"
+    assert len(describe_response["BlueGreenDeployments"]) == 0
+    assert len(describe_source_instance["DBInstances"]) == 1
+    if options["delete_target"] and not options["with_switchover"]:
+        assert len(describe_target_instance["DBInstances"]) == 0
+    else:
+        assert len(describe_target_instance["DBInstances"]) == 1
+
+
+@mock_aws
+def test_delete_blue_green_deployment_when_it_does_not_exist(client):
+    with pytest.raises(ClientError) as exc:
+        client.delete_blue_green_deployment(
+            BlueGreenDeploymentIdentifier="DoesNotExist"
+        )
+
+    error = exc.value.response["Error"]
+    assert error["Code"] == "BlueGreenDeploymentNotFoundFault"
+
+
+@mock_aws
+@pytest.mark.parametrize(
+    "options",
+    [
+        pytest.param(
+            {"with_switchover": False, "delete_target": True},
+            id="before_switchover_with_delete_target",
+        ),
+        pytest.param(
+            {"with_switchover": False, "delete_target": False},
+            id="before_switchover_without_delete_target",
+        ),
+        pytest.param(
+            {"with_switchover": True, "delete_target": True},
+            id="after_switchover_with_delete_target",
+        ),
+    ],
+)
+def test_delete_blue_green_deployment_with_source_db_cluster(client, options):
+    bg_name = "bluegreen-deployment-1"
+
+    create_db_cluster(
+        DBClusterIdentifier="cluster-1",
+        Engine="aurora-postgresql",
+    )
+    clustered_instances = []
+    for i in range(3):
+        clustered_instances.append(
+            client.create_db_instance(
+                DBInstanceIdentifier=f"test-instance-{i}",
+                DBClusterIdentifier="cluster-1",
+                DBInstanceClass="db.m1.small",
+                Engine="aurora-postgresql",
+            )
+        )
+    cluster = client.describe_db_clusters(
+        DBClusterIdentifier="cluster-1",
+    )["DBClusters"][0]
+
+    create_response = client.create_blue_green_deployment(
+        BlueGreenDeploymentName=bg_name,
+        Source=cluster["DBClusterArn"],
+    )
+
+    if options["with_switchover"]:
+        client.switchover_blue_green_deployment(
+            BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+                "BlueGreenDeploymentIdentifier"
+            ]
+        )
+
+    delete_response = client.delete_blue_green_deployment(
+        BlueGreenDeploymentIdentifier=create_response["BlueGreenDeployment"][
+            "BlueGreenDeploymentIdentifier"
+        ],
+        DeleteTarget=options["delete_target"],
+    )
+
+    describe_response = client.describe_blue_green_deployments(
+        Filters=[{"Name": "blue-green-deployment-name", "Values": [bg_name]}]
+    )
+
+    describe_source_cluster = client.describe_db_clusters(
+        DBClusterIdentifier=delete_response["BlueGreenDeployment"]["Source"]
+    )
+
+    describe_target_cluster = client.describe_db_clusters(
+        Filters=[
+            {
+                "Name": "db-cluster-id",
+                "Values": [delete_response["BlueGreenDeployment"]["Target"]],
+            }
+        ]
+    )
+
+    assert delete_response["BlueGreenDeployment"]["Status"] == "DELETING"
+    assert len(describe_response["BlueGreenDeployments"]) == 0
+    assert len(describe_source_cluster["DBClusters"]) == 1
+    if options["delete_target"] and not options["with_switchover"]:
+        assert len(describe_target_cluster["DBClusters"]) == 0
+    else:
+        assert len(describe_target_cluster["DBClusters"]) == 1
+
+
+def update_status_from_create_blue_green_response(bg_response: dict) -> None:
+    bg_response["Status"] = "AVAILABLE"
+
+    for details in bg_response["SwitchoverDetails"]:
+        details["Status"] = "AVAILABLE"
+    for task in bg_response["Tasks"]:
+        task["Status"] = "COMPLETED"
diff --git a/tests/test_s3/test_s3_content_encoding.py b/tests/test_s3/test_s3_content_encoding.py
new file mode 100644
index 000000000..7ebe6decf
--- /dev/null
+++ b/tests/test_s3/test_s3_content_encoding.py
@@ -0,0 +1,30 @@
+import os
+from tempfile import TemporaryFile
+
+import boto3
+import pytest
+
+from . import s3_aws_verified
+
+
+@pytest.mark.aws_verified
+@s3_aws_verified
+@pytest.mark.parametrize("encoding", [None, "gzip", "unknown"])
+def test_content_encoding_is_returned(encoding, bucket_name=None):
+    client = boto3.client("s3", region_name="us-east-1")
+    s3_file = "some.file"
+
+    attrs = {
+        "ContentType": "application/octet-stream",
+        "ACL": "bucket-owner-full-control",
+    }
+    if encoding:
+        attrs["ContentEncoding"] = encoding
+
+    with TemporaryFile() as f:
+        f.write(os.urandom(1024))
+        f.flush()
+        client.upload_fileobj(f, bucket_name, s3_file, ExtraArgs=attrs)
+
+    res = client.get_object(Bucket=bucket_name, Key=s3_file)
+    assert res.get("ContentEncoding") == encoding
diff --git a/tests/test_secretsmanager/test_secrets_duplication.py b/tests/test_secretsmanager/test_secrets_duplication.py
index 1f7b1988b..c343f9838 100644
--- a/tests/test_secretsmanager/test_secrets_duplication.py
+++ b/tests/test_secretsmanager/test_secrets_duplication.py
@@ -97,6 +97,42 @@ def test_replicate_secret():
     assert replica_secret["ARN"] == replica_arn
 
 
+@mock_aws
+def test_list_replica_secret_with_filters():
+    us1 = boto3.client("secretsmanager", region_name="us-east-1")
+    us2 = boto3.client("secretsmanager", region_name="us-east-2")
+
+    resp = us1.create_secret(
+        Name="dup",
+        SecretString="s",
+        Tags=[
+            {"Key": "k", "Value": "v"},
+            {
+                # coercing 'owning-service' filter to trigger
+                "Key": "aws:secretsmanager:owningService",
+                "Value": "testing-only",
+            },
+        ],
+        Description="description",
+    )
+    primary_arn = resp["ARN"]
+
+    us1.replicate_secret_to_regions(
+        SecretId=primary_arn, AddReplicaRegions=[{"Region": "us-east-2"}]
+    )
+
+    filters = [
+        {"Key": "tag-key", "Values": ["k"]},
+        {"Key": "tag-value", "Values": ["v"]},
+        {"Key": "all", "Values": ["k"]},
+        {"Key": "description", "Values": ["description"]},
+        {"Key": "owning-service", "Values": ["testing-only"]},
+    ]
+
+    for f in filters:
+        assert len(us2.list_secrets(Filters=[f])["SecretList"]) == 1
+
+
 @mock_aws
 def test_replicate_secret__in_same_region():
     conn = boto3.client("secretsmanager", region_name="us-east-1")