Repository: getmoto/moto (python)

Issue:
S3 list bucket and list all buckets response should have content-type application/xml instead of text/html
Currently a lot of responses incorrectly have content-type text/html.  Instead they should have application/xml when appropriate.

Pull request:
S3: Add Content-Type=application/xml headers where appropriate
Fixes #8560 

Patch:
diff --git a/moto/core/exceptions.py b/moto/core/exceptions.py
index 94f9b8189..11d886f33 100644
--- a/moto/core/exceptions.py
+++ b/moto/core/exceptions.py
@@ -4,6 +4,8 @@ from typing import Any, Dict, List, Optional, Tuple
 from jinja2 import DictLoader, Environment
 from werkzeug.exceptions import HTTPException
 
+from moto.core.mime_types import APP_XML
+
 
 class ServiceException(Exception):
     """
@@ -117,7 +119,7 @@ class RESTError(HTTPException):
                 include_type_sender=self.include_type_sender,
                 **kwargs,
             )
-            self.content_type = "application/xml"
+            self.content_type = APP_XML
 
     def get_headers(
         self,
diff --git a/moto/core/mime_types.py b/moto/core/mime_types.py
new file mode 100644
index 000000000..8a7211ced
--- /dev/null
+++ b/moto/core/mime_types.py
@@ -0,0 +1,1 @@
+APP_XML = "application/xml"
diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index f27b1589a..aa527bc99 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -10,6 +10,7 @@ import xmltodict
 
 from moto import settings
 from moto.core.common_types import TYPE_RESPONSE
+from moto.core.mime_types import APP_XML
 from moto.core.responses import BaseResponse
 from moto.core.utils import (
     ALT_DOMAIN_SUFFIXES,
@@ -338,15 +339,19 @@ class S3Response(BaseResponse):
         return self._send_response(response)
 
     @staticmethod
-    def _send_response(response: Any) -> TYPE_RESPONSE:  # type: ignore
-        if isinstance(response, str):
-            return 200, {}, response.encode("utf-8")
+    def _send_response(response: Union[TYPE_RESPONSE, str, bytes]) -> TYPE_RESPONSE:  # type: ignore
+        if isinstance(response, (str, bytes)):
+            status_code = 200
+            headers: dict[str, Any] = {}
         else:
-            status_code, headers, response_content = response
-            if not isinstance(response_content, bytes):
-                response_content = response_content.encode("utf-8")
+            status_code, headers, response = response
+        if not isinstance(response, bytes):
+            response = response.encode("utf-8")
+
+        if response and "content-type" not in headers:
+            headers["content-type"] = APP_XML
 
-            return status_code, headers, response_content
+        return status_code, headers, response
 
     def _bucket_response(
         self, request: Any, full_url: str
@@ -411,7 +416,8 @@ class S3Response(BaseResponse):
             # raises NoSuchBucket, leading to inconsistency in
             # error response between real and mocked responses.
             return 404, {}, ""
-        return 200, {"x-amz-bucket-region": bucket.region_name}, ""
+        headers = {"x-amz-bucket-region": bucket.region_name, "content-type": APP_XML}
+        return 200, headers, ""
 
     def _set_cors_headers_options(
         self, headers: Dict[str, str], bucket: FakeBucket
@@ -1336,7 +1342,7 @@ class S3Response(BaseResponse):
 
         if (begin is None and end == 0) or (begin is not None and begin > last):
             if request.method == "HEAD":
-                return 416, {"Content-Type": "application/xml"}, b""
+                return 416, {"Content-Type": APP_XML}, b""
             raise InvalidRange(
                 actual_size=str(length), range_requested=request.headers.get("range")
             )
@@ -1409,18 +1415,13 @@ class S3Response(BaseResponse):
         self.setup_class(request, full_url, headers)
         bucket_name = self.parse_bucket_name_from_url(request, full_url)
         self.backend.log_incoming_request(request, bucket_name)
-        response_headers: Dict[str, Any] = {}
 
         try:
             response = self._key_response(request, full_url)
         except S3ClientError as s3error:
             response = s3error.code, {}, s3error.description
 
-        if isinstance(response, str):
-            status_code = 200
-            response_content = response
-        else:
-            status_code, response_headers, response_content = response
+        status_code, response_headers, response_content = self._send_response(response)
 
         if (
             status_code == 200
@@ -2109,7 +2110,7 @@ class S3Response(BaseResponse):
             headers = {
                 "x-amz-delete-marker": "true",
                 "x-amz-version-id": version_id,
-                "content-type": "application/xml",
+                "content-type": APP_XML,
             }
             if version_id:
                 headers["allow"] = "DELETE"


Test patch:
diff --git a/tests/test_s3/test_s3_response_headers.py b/tests/test_s3/test_s3_response_headers.py
new file mode 100644
index 000000000..e5588c4af
--- /dev/null
+++ b/tests/test_s3/test_s3_response_headers.py
@@ -0,0 +1,32 @@
+import boto3
+import pytest
+
+from . import s3_aws_verified
+from .test_s3 import DEFAULT_REGION_NAME
+
+
+@pytest.mark.aws_verified
+@s3_aws_verified
+def test_header_content_type(bucket_name=None):
+    client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
+
+    resp = client.list_buckets()
+    headers = resp["ResponseMetadata"]["HTTPHeaders"]
+    assert headers.get("content-type") == "application/xml"
+
+    resp = client.list_object_versions(Bucket=bucket_name)
+    headers = resp["ResponseMetadata"]["HTTPHeaders"]
+    assert headers.get("content-type") == "application/xml"
+
+    resp = client.head_bucket(Bucket=bucket_name)
+    headers = resp["ResponseMetadata"]["HTTPHeaders"]
+    assert headers.get("content-type") == "application/xml"
+
+    resp = client.put_object(Bucket=bucket_name, Key="key", Body=b"data")
+    headers = resp["ResponseMetadata"]["HTTPHeaders"]
+    # Werkzeug automatically adds a content-type if none is specified
+    assert headers.get("content-type") in [None, "text/html; charset=utf-8"]
+
+    resp = client.get_object(Bucket=bucket_name, Key="key")
+    headers = resp["ResponseMetadata"]["HTTPHeaders"]
+    assert headers.get("content-type") == "binary/octet-stream"