Repository: getmoto/moto
Pull request #9004 (2025-06-22 19:45:48)

RAM - Some refactoring for the next steps
Hello, again!

In this PR, I'm implementing some refatoring in the `models.py` file. Since I'm willing to implement new logic to interact with the managed permissions specifications, I decided to refactor the models we have and introduce two new model classes, in order to configure the backend for the next RAM implementations. Here's a summary of the changes:
- I created the `ResourceType` class to represent a model of any resource type implemented in the backend of the RAM service;
- I created the `ManagedPermission` class to represent a model of any managed permission implemented in the backend of the RAM service;
- Since I introduced two new classes to represent parts of the service, I added new class attributes for the `ResourceAccessManagerBackend` class. The difference now is that each attribute we had previously now represents a `List` of objects from the classes I mentioned above;
- Since I created the new attributes above, I had to refactor the previously implemented methods in order to work with this new pattern;
- I refactored the `test_ram.py` file, to make use of the `AWS_MANAGED_PERMISSIONS` and `RAM_RESOURCE_TYPES` lists for validation;
- I created the `ManagedPermissionDict` in `utils.py` for better type annotation and removed the unnecessary `str()` calls from `models.py` (thank you @bblommers for the recommendation);
- Lastly, I did some minor changes I noticed by re-reading the `models.py` file (e.g. typo in the `resouce` variable name in `organizations_backend()` method)

I'd like to discuss if this new pattern implementation for the service makes sense. Thanks in advance for the help :)


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/ram/models.py b/moto/ram/models.py
index c8c15e642..dbf739ef1 100644
--- a/moto/ram/models.py
+++ b/moto/ram/models.py
@@ -1,12 +1,16 @@
 import re
 import string
+from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
 from moto.core.base_backend import BackendDict, BaseBackend
 from moto.core.common_models import BaseModel
 from moto.core.utils import unix_time, utcnow
 from moto.moto_api._internal import mock_random as random
-from moto.organizations.models import OrganizationsBackend, organizations_backends
+from moto.organizations.models import (
+    OrganizationsBackend,
+    organizations_backends,
+)
 from moto.ram.exceptions import (
     InvalidParameterException,
     MalformedArnException,
@@ -50,7 +54,7 @@ class ResourceShare(BaseModel):
         self.partition = get_partition(region)
 
         self.allow_external_principals = kwargs.get("allowExternalPrincipals", True)
-        self.arn = f"arn:{get_partition(self.region)}:ram:{self.region}:{account_id}:resource-share/{random.uuid4()}"
+        self.arn = f"arn:{self.partition}:ram:{self.region}:{account_id}:resource-share/{random.uuid4()}"
         self.creation_time = utcnow()
         self.feature_set = "STANDARD"
         self.last_updated_time = utcnow()
@@ -154,16 +158,104 @@ class ResourceShare(BaseModel):
         self.name = name
 
 
-class ResourceAccessManagerBackend(BaseBackend):
-    RESOURCE_TYPES = RAM_RESOURCE_TYPES
+class ResourceType(BaseModel):
+    def __init__(
+        self, resource_type: str, service_name: str, resource_region_scope: str
+    ):
+        self.resource_type = resource_type
+        self.service_name = service_name
+        self.resource_region_scope = resource_region_scope
 
-    PERMISSION_TYPES = ["ALL", "AWS", "LOCAL"]
+    def describe(self) -> Dict[str, Any]:
+        return {
+            "resourceType": self.resource_type,
+            "serviceName": self.service_name,
+            "resourceRegionScope": self.resource_region_scope,
+        }
+
+
+class ManagedPermission(BaseModel):
+    def __init__(
+        self,
+        account_id: str,
+        region: str,
+        name: str,
+        resource_type: str,
+        version: str = "1",
+        default_version: bool = True,
+        status: str = "ATTACHABLE",
+        creation_time: Optional[str] = None,
+        last_updated_time: Optional[str] = None,
+        is_resource_type_default: bool = False,
+        permission_type: str = "AWS_MANAGED",  # or "CUSTOMER_MANAGED",
+    ):
+        self.account_id = account_id
+        self.region = region
+        self.partition = get_partition(region)
+
+        now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
+        arn_prefix = (
+            f"arn:{self.partition}:ram::{self.partition}:permission/"
+            if permission_type == "AWS_MANAGED"
+            else f"arn:{self.partition}:ram:{self.region}:{account_id}:permission/"
+        )
+
+        self.name = name
+        self.arn = f"{arn_prefix}{name}"
+        self.resource_type = resource_type
+        self.version = version
+        self.default_version = default_version
+        self.status = status
+        self.creation_time = creation_time or now
+        self.last_updated_time = last_updated_time or creation_time
+        self.is_resource_type_default = is_resource_type_default
+        self.permission_type = permission_type
 
-    MANAGED_PERMISSIONS = AWS_MANAGED_PERMISSIONS
+    def describe(self) -> Dict[str, Any]:
+        return {
+            "arn": self.arn,
+            "name": self.name,
+            "resourceType": self.resource_type,
+            "version": self.version,
+            "defaultVersion": self.default_version,
+            "status": self.status,
+            "creationTime": self.creation_time,
+            "lastUpdatedTime": self.last_updated_time,
+            "isResourceTypeDefault": self.is_resource_type_default,
+            "permissionType": self.permission_type,
+        }
+
+
+class ResourceAccessManagerBackend(BaseBackend):
+    PERMISSION_TYPES = ["ALL", "AWS", "LOCAL"]
 
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.resource_shares: List[ResourceShare] = []
+        self.managed_permissions: List[ManagedPermission] = [
+            ManagedPermission(
+                account_id=account_id,
+                region=region_name,
+                name=permission["name"],
+                resource_type=permission["resourceType"],
+                version=permission["version"],
+                default_version=permission["defaultVersion"],
+                status=permission["status"],
+                creation_time=permission["creationTime"],
+                last_updated_time=permission["lastUpdatedTime"],
+                is_resource_type_default=permission["isResourceTypeDefault"],
+                permission_type=permission["permissionType"],
+            )
+            for permission in AWS_MANAGED_PERMISSIONS
+        ]
+        self.resource_types: List[ResourceType] = [
+            ResourceType(
+                resource_type=resource_type["resourceType"],
+                service_name=resource_type["serviceName"],
+                resource_region_scope=resource_type["resourceRegionScope"],
+            )
+            for resource_type in RAM_RESOURCE_TYPES
+        ]
 
     @property
     def organizations_backend(self) -> OrganizationsBackend:
@@ -193,9 +285,9 @@ class ResourceAccessManagerBackend(BaseBackend):
                 "Value 'OTHER-ACCOUNTS' for parameter 'resourceOwner' not implemented."
             )
 
-        resouces = [resource.describe() for resource in self.resource_shares]
+        resources = [resource.describe() for resource in self.resource_shares]
 
-        return dict(resourceShares=resouces)
+        return dict(resourceShares=resources)
 
     def update_resource_share(
         self,
@@ -336,12 +428,15 @@ class ResourceAccessManagerBackend(BaseBackend):
             )
 
         if resource_region_scope == "ALL":
-            resource_types = self.RESOURCE_TYPES
+            resource_types = [
+                resource_type.describe() for resource_type in self.resource_types
+            ]
         else:
             resource_types = [
-                resource_type
-                for resource_type in self.RESOURCE_TYPES
-                if resource_type["resourceRegionScope"] == resource_region_scope
+                resource_type_dict
+                for resource_type in self.resource_types
+                if (resource_type_dict := resource_type.describe())
+                and resource_type_dict["resourceRegionScope"] == resource_region_scope
             ]
 
         return dict(resourceTypes=resource_types)
@@ -357,20 +452,25 @@ class ResourceAccessManagerBackend(BaseBackend):
         # Here, resourceType first partition (service) is case sensitive and
         # last partition (type) is case insensitive
         if resource_type and not any(
-            str(permission["resourceType"]).split(":")[0] == resource_type.split(":")[0]
-            and str(permission["resourceType"]).lower() == resource_type.lower()
-            for permission in self.MANAGED_PERMISSIONS
+            (permission_dict := permission.describe())
+            and permission_dict["resourceType"].split(":")[0]
+            == resource_type.split(":")[0]
+            and permission_dict["resourceType"].lower() == resource_type.lower()
+            for permission in self.managed_permissions
         ):
             raise InvalidParameterException(f"Invalid resource type: {resource_type}")
 
         if resource_type:
             permissions = [
-                permission
-                for permission in self.MANAGED_PERMISSIONS
-                if str(permission["resourceType"]).lower() == resource_type.lower()
+                permission_dict
+                for permission in self.managed_permissions
+                if (permission_dict := permission.describe())
+                and permission_dict["resourceType"].lower() == resource_type.lower()
             ]
         else:
-            permissions = self.MANAGED_PERMISSIONS
+            permissions = [
+                permission.describe() for permission in self.managed_permissions
+            ]
 
         if permission_type not in self.PERMISSION_TYPES:
             raise InvalidParameterException(
diff --git a/moto/ram/utils.py b/moto/ram/utils.py
index 1de1e0c05..5e78d4685 100644
--- a/moto/ram/utils.py
+++ b/moto/ram/utils.py
@@ -1,5 +1,18 @@
 from datetime import datetime, timezone
-from typing import Dict, Optional, Union
+from typing import Optional, TypedDict
+
+
+class ManagedPermissionDict(TypedDict):
+    arn: str
+    version: str
+    defaultVersion: bool
+    name: str
+    resourceType: str
+    status: str
+    creationTime: str
+    lastUpdatedTime: str
+    isResourceTypeDefault: bool
+    permissionType: str
 
 
 def format_ram_permission(
@@ -13,7 +26,7 @@ def format_ram_permission(
     is_resource_type_default: bool = True,
     permission_type: str = "AWS_MANAGED",
     default_version: bool = True,
-) -> Dict[str, Union[str, bool]]:
+) -> ManagedPermissionDict:
     """
     Format a RAM (Resource Access Manager) permission dictionary with the
     specified attributes.
@@ -38,7 +51,7 @@ def format_ram_permission(
             Defaults to True.
 
     Returns:
-        dict: A dictionary representing the formatted RAM permission.
+        ManagedPermissionDict: A dictionary representing the formatted RAM permission.
     """
     now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
     creation_time = creation_time or now
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:


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_ram/test_ram.py b/tests/test_ram/test_ram.py
index d3637c644..d161b4a97 100644
--- a/tests/test_ram/test_ram.py
+++ b/tests/test_ram/test_ram.py
@@ -9,7 +9,7 @@ from botocore.exceptions import ClientError
 
 from moto import mock_aws
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
-from moto.ram.models import ResourceAccessManagerBackend
+from moto.ram.utils import AWS_MANAGED_PERMISSIONS, RAM_RESOURCE_TYPES
 
 
 @mock_aws
@@ -569,7 +569,7 @@ def test_list_resource_types(resource_region_scope, expect_error, error_message)
         assert ex.response["Error"]["Message"] == error_message
     else:
         response = client.list_resource_types(**resource_region_scope)
-        expected_types = ResourceAccessManagerBackend.RESOURCE_TYPES
+        expected_types = RAM_RESOURCE_TYPES
         if region_scope == "GLOBAL":
             expected_types = [
                 rt for rt in expected_types if rt["resourceRegionScope"] == "GLOBAL"
@@ -632,7 +632,7 @@ def test_list_permissions(parameters, expect_error, error_message):
         response = client.list_permissions(**parameters)
 
         # then
-        expected_permissions = ResourceAccessManagerBackend.MANAGED_PERMISSIONS
+        expected_permissions = AWS_MANAGED_PERMISSIONS
         if resource_type:
             expected_permissions = [
                 permission
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