Fix REST sensor charset handling to respect Content-Type header (home-assistant#148223)

Author: bdraco

homeassistant/components/rest/data.py

@@ -140,7 +140,14 @@ async def async_update(self, log_errors: bool = True) -> None:
                 self._method, self._resource, **request_kwargs
             ) as response:
                 # Read the response
-                self.data = await response.text(encoding=self._encoding)
+                # Only use configured encoding if no charset in Content-Type header
+                # If charset is present in Content-Type, let aiohttp use it
+                if response.charset:
+                    # Let aiohttp use the charset from Content-Type header
+                    self.data = await response.text()
+                else:
+                    # Use configured encoding as fallback
+                    self.data = await response.text(encoding=self._encoding)
                 self.headers = response.headers
 
         except TimeoutError as ex:

tests/components/rest/test_sensor.py

@@ -162,6 +162,94 @@ async def test_setup_encoding(
     assert hass.states.get("sensor.mysensor").state == "tack själv"
 
 
+async def test_setup_auto_encoding_from_content_type(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test setup with encoding auto-detected from Content-Type header."""
+    # Test with ISO-8859-1 charset in Content-Type header
+    aioclient_mock.get(
+        "http://localhost",
+        status=HTTPStatus.OK,
+        content="Björk Guðmundsdóttir".encode("iso-8859-1"),
+        headers={"Content-Type": "text/plain; charset=iso-8859-1"},
+    )
+    assert await async_setup_component(
+        hass,
+        SENSOR_DOMAIN,
+        {
+            SENSOR_DOMAIN: {
+                "name": "mysensor",
+                # encoding defaults to UTF-8, but should be ignored when charset present
+                "platform": DOMAIN,
+                "resource": "http://localhost",
+                "method": "GET",
+            }
+        },
+    )
+    await hass.async_block_till_done()
+    assert len(hass.states.async_all(SENSOR_DOMAIN)) == 1
+    assert hass.states.get("sensor.mysensor").state == "Björk Guðmundsdóttir"
+
+
+async def test_setup_encoding_fallback_no_charset(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test that configured encoding is used when no charset in Content-Type."""
+    # No charset in Content-Type header
+    aioclient_mock.get(
+        "http://localhost",
+        status=HTTPStatus.OK,
+        content="Björk Guðmundsdóttir".encode("iso-8859-1"),
+        headers={"Content-Type": "text/plain"},  # No charset!
+    )
+    assert await async_setup_component(
+        hass,
+        SENSOR_DOMAIN,
+        {
+            SENSOR_DOMAIN: {
+                "name": "mysensor",
+                "encoding": "iso-8859-1",  # This will be used as fallback
+                "platform": DOMAIN,
+                "resource": "http://localhost",
+                "method": "GET",
+            }
+        },
+    )
+    await hass.async_block_till_done()
+    assert len(hass.states.async_all(SENSOR_DOMAIN)) == 1
+    assert hass.states.get("sensor.mysensor").state == "Björk Guðmundsdóttir"
+
+
+async def test_setup_charset_overrides_encoding_config(
+    hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
+) -> None:
+    """Test that charset in Content-Type overrides configured encoding."""
+    # Server sends UTF-8 with correct charset header
+    aioclient_mock.get(
+        "http://localhost",
+        status=HTTPStatus.OK,
+        content="Björk Guðmundsdóttir".encode(),
+        headers={"Content-Type": "text/plain; charset=utf-8"},
+    )
+    assert await async_setup_component(
+        hass,
+        SENSOR_DOMAIN,
+        {
+            SENSOR_DOMAIN: {
+                "name": "mysensor",
+                "encoding": "iso-8859-1",  # Config says ISO-8859-1, but charset=utf-8 should win
+                "platform": DOMAIN,
+                "resource": "http://localhost",
+                "method": "GET",
+            }
+        },
+    )
+    await hass.async_block_till_done()
+    assert len(hass.states.async_all(SENSOR_DOMAIN)) == 1
+    # This should work because charset=utf-8 overrides the iso-8859-1 config
+    assert hass.states.get("sensor.mysensor").state == "Björk Guðmundsdóttir"
+
+
 @pytest.mark.parametrize(
     ("ssl_cipher_list", "ssl_cipher_list_expected"),
     [

tests/test_util/aiohttp.py

@@ -191,7 +191,6 @@ def __init__(
         if response is None:
             response = b""
 
-        self.charset = "utf-8"
         self.method = method
         self._url = url
         self.status = status
@@ -261,16 +260,32 @@ def content(self):
         """Return content."""
         return mock_stream(self.response)
 
+    @property
+    def charset(self):
+        """Return charset from Content-Type header."""
+        if (content_type := self._headers.get("content-type")) is None:
+            return None
+        content_type = content_type.lower()
+        if "charset=" in content_type:
+            return content_type.split("charset=")[1].split(";")[0].strip()
+        return None
+
     async def read(self):
         """Return mock response."""
         return self.response
 
-    async def text(self, encoding="utf-8", errors="strict"):
+    async def text(self, encoding=None, errors="strict") -> str:
         """Return mock response as a string."""
+        # Match real aiohttp behavior: encoding=None means auto-detect
+        if encoding is None:
+            encoding = self.charset or "utf-8"
         return self.response.decode(encoding, errors=errors)
 
-    async def json(self, encoding="utf-8", content_type=None, loads=json_loads):
+    async def json(self, encoding=None, content_type=None, loads=json_loads) -> Any:
         """Return mock response as a json."""
+        # Match real aiohttp behavior: encoding=None means auto-detect
+        if encoding is None:
+            encoding = self.charset or "utf-8"
         return loads(self.response.decode(encoding))
 
     def release(self):