Add image entity component (#90564)

.core_files.yaml

@@ -27,6 +27,7 @@ base_platforms: &base_platforms
   - homeassistant/components/fan/**
   - homeassistant/components/geo_location/**
   - homeassistant/components/humidifier/**
+  - homeassistant/components/image/**
   - homeassistant/components/image_processing/**
   - homeassistant/components/light/**
   - homeassistant/components/lock/**

.strict-typing

@@ -172,6 +172,7 @@ homeassistant.components.huawei_lte.*
 homeassistant.components.hydrawise.*
 homeassistant.components.hyperion.*
 homeassistant.components.ibeacon.*
+homeassistant.components.image.*
 homeassistant.components.image_processing.*
 homeassistant.components.image_upload.*
 homeassistant.components.imap.*

CODEOWNERS

@@ -563,6 +563,8 @@ build.json @home-assistant/supervisor
 /tests/components/icloud/ @Quentame @nzapponi
 /homeassistant/components/ign_sismologia/ @exxamalte
 /tests/components/ign_sismologia/ @exxamalte
+/homeassistant/components/image/ @home-assistant/core
+/tests/components/image/ @home-assistant/core
 /homeassistant/components/image_processing/ @home-assistant/core
 /tests/components/image_processing/ @home-assistant/core
 /homeassistant/components/image_upload/ @home-assistant/core

homeassistant/components/image/__init__.py

@@ -0,0 +1,211 @@
+"""The image integration."""
+from __future__ import annotations
+
+import asyncio
+import collections
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+import logging
+from random import SystemRandom
+from typing import Final, final
+
+from aiohttp import hdrs, web
+import async_timeout
+
+from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP
+from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.config_validation import (  # noqa: F401
+    PLATFORM_SCHEMA,
+    PLATFORM_SCHEMA_BASE,
+)
+from homeassistant.helpers.entity import Entity, EntityDescription
+from homeassistant.helpers.entity_component import EntityComponent
+from homeassistant.helpers.event import async_track_time_interval
+from homeassistant.helpers.typing import ConfigType
+
+from .const import DOMAIN, IMAGE_TIMEOUT  # noqa: F401
+
+_LOGGER = logging.getLogger(__name__)
+
+SCAN_INTERVAL: Final = timedelta(seconds=30)
+ENTITY_ID_FORMAT: Final = DOMAIN + ".{}"
+
+DEFAULT_CONTENT_TYPE: Final = "image/jpeg"
+ENTITY_IMAGE_URL: Final = "/api/image_proxy/{0}?token={1}"
+
+TOKEN_CHANGE_INTERVAL: Final = timedelta(minutes=5)
+_RND: Final = SystemRandom()
+
+
+@dataclass
+class ImageEntityDescription(EntityDescription):
+    """A class that describes image entities."""
+
+
+@dataclass
+class Image:
+    """Represent an image."""
+
+    content_type: str
+    content: bytes
+
+
+async def _async_get_image(image_entity: ImageEntity, timeout: int) -> Image:
+    """Fetch image from an image entity."""
+    with suppress(asyncio.CancelledError, asyncio.TimeoutError):
+        async with async_timeout.timeout(timeout):
+            if image_bytes := await image_entity.async_image():
+                content_type = image_entity.content_type
+                image = Image(content_type, image_bytes)
+                return image
+
+    raise HomeAssistantError("Unable to get image")
+
+
+async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+    """Set up the image component."""
+    component = hass.data[DOMAIN] = EntityComponent[ImageEntity](
+        _LOGGER, DOMAIN, hass, SCAN_INTERVAL
+    )
+
+    hass.http.register_view(ImageView(component))
+
+    await component.async_setup(config)
+
+    @callback
+    def update_tokens(time: datetime) -> None:
+        """Update tokens of the entities."""
+        for entity in component.entities:
+            entity.async_update_token()
+            entity.async_write_ha_state()
+
+    unsub = async_track_time_interval(
+        hass, update_tokens, TOKEN_CHANGE_INTERVAL, name="Image update tokens"
+    )
+
+    @callback
+    def unsub_track_time_interval(_event: Event) -> None:
+        """Unsubscribe track time interval timer."""
+        unsub()
+
+    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, unsub_track_time_interval)
+
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up a config entry."""
+    component: EntityComponent[ImageEntity] = hass.data[DOMAIN]
+    return await component.async_setup_entry(entry)
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    component: EntityComponent[ImageEntity] = hass.data[DOMAIN]
+    return await component.async_unload_entry(entry)
+
+
+class ImageEntity(Entity):
+    """The base class for image entities."""
+
+    # Entity Properties
+    _attr_content_type: str = DEFAULT_CONTENT_TYPE
+    _attr_image_last_updated: datetime | None = None
+    _attr_should_poll: bool = False  # No need to poll image entities
+    _attr_state: None = None  # State is determined by last_updated
+
+    def __init__(self) -> None:
+        """Initialize an image entity."""
+        self.access_tokens: collections.deque = collections.deque([], 2)
+        self.async_update_token()
+
+    @property
+    def content_type(self) -> str:
+        """Image content type."""
+        return self._attr_content_type
+
+    @property
+    def entity_picture(self) -> str:
+        """Return a link to the image as entity picture."""
+        if self._attr_entity_picture is not None:
+            return self._attr_entity_picture
+        return ENTITY_IMAGE_URL.format(self.entity_id, self.access_tokens[-1])
+
+    @property
+    def image_last_updated(self) -> datetime | None:
+        """The time when the image was last updated."""
+        return self._attr_image_last_updated
+
+    def image(self) -> bytes | None:
+        """Return bytes of image."""
+        raise NotImplementedError()
+
+    async def async_image(self) -> bytes | None:
+        """Return bytes of image."""
+        return await self.hass.async_add_executor_job(self.image)
+
+    @property
+    @final
+    def state(self) -> str | None:
+        """Return the state."""
+        if self.image_last_updated is None:
+            return None
+        return self.image_last_updated.isoformat()
+
+    @final
+    @property
+    def state_attributes(self) -> dict[str, str | None]:
+        """Return the state attributes."""
+        return {"access_token": self.access_tokens[-1]}
+
+    @callback
+    def async_update_token(self) -> None:
+        """Update the used token."""
+        self.access_tokens.append(hex(_RND.getrandbits(256))[2:])
+
+
+class ImageView(HomeAssistantView):
+    """View to serve an image."""
+
+    name = "api:image:image"
+    requires_auth = False
+    url = "/api/image_proxy/{entity_id}"
+
+    def __init__(self, component: EntityComponent[ImageEntity]) -> None:
+        """Initialize an image view."""
+        self.component = component
+
+    async def get(self, request: web.Request, entity_id: str) -> web.StreamResponse:
+        """Start a GET request."""
+        if (image_entity := self.component.get_entity(entity_id)) is None:
+            raise web.HTTPNotFound()
+
+        authenticated = (
+            request[KEY_AUTHENTICATED]
+            or request.query.get("token") in image_entity.access_tokens
+        )
+
+        if not authenticated:
+            # Attempt with invalid bearer token, raise unauthorized
+            # so ban middleware can handle it.
+            if hdrs.AUTHORIZATION in request.headers:
+                raise web.HTTPUnauthorized()
+            # Invalid sigAuth or image entity access token
+            raise web.HTTPForbidden()
+
+        return await self.handle(request, image_entity)
+
+    async def handle(
+        self, request: web.Request, image_entity: ImageEntity
+    ) -> web.StreamResponse:
+        """Serve image."""
+        try:
+            image = await _async_get_image(image_entity, IMAGE_TIMEOUT)
+        except (HomeAssistantError, ValueError) as ex:
+            raise web.HTTPInternalServerError() from ex
+
+        return web.Response(body=image.content, content_type=image.content_type)

homeassistant/components/image/const.py

@@ -0,0 +1,6 @@
+"""Constants for the image integration."""
+from typing import Final
+
+DOMAIN: Final = "image"
+
+IMAGE_TIMEOUT: Final = 10

homeassistant/components/image/manifest.json

@@ -0,0 +1,9 @@
+{
+  "domain": "image",
+  "name": "Image",
+  "codeowners": ["@home-assistant/core"],
+  "dependencies": ["http"],
+  "documentation": "https://www.home-assistant.io/integrations/image",
+  "integration_type": "entity",
+  "quality_scale": "internal"
+}

homeassistant/components/image/recorder.py

@@ -0,0 +1,10 @@
+"""Integration platform for recorder."""
+from __future__ import annotations
+
+from homeassistant.core import HomeAssistant, callback
+
+
+@callback
+def exclude_attributes(hass: HomeAssistant) -> set[str]:
+    """Exclude access_token and entity_picture from being recorded in the database."""
+    return {"access_token", "entity_picture"}

homeassistant/components/image/strings.json

@@ -0,0 +1,8 @@
+{
+  "title": "Image",
+  "entity_component": {
+    "_": {
+      "name": "[%key:component::image::title%]"
+    }
+  }
+}

homeassistant/components/kitchen_sink/__init__.py

@@ -26,7 +26,7 @@ import homeassistant.util.dt as dt_util
 DOMAIN = "kitchen_sink"
 
 
-COMPONENTS_WITH_DEMO_PLATFORM = [Platform.SENSOR, Platform.LOCK]
+COMPONENTS_WITH_DEMO_PLATFORM = [Platform.SENSOR, Platform.LOCK, Platform.IMAGE]
 
 CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
 

homeassistant/components/kitchen_sink/image.py

@@ -0,0 +1,66 @@
+"""Demo image platform."""
+from __future__ import annotations
+
+from pathlib import Path
+
+from homeassistant.components.image import ImageEntity
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.util import dt as dt_util
+
+
+async def async_setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    async_add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None,
+) -> None:
+    """Set up image entities."""
+    async_add_entities(
+        [
+            DemoImage(
+                "kitchen_sink_image_001",
+                "QR Code",
+                "image/png",
+                "qr_code.png",
+            ),
+        ]
+    )
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the Everything but the Kitchen Sink config entry."""
+    await async_setup_platform(hass, {}, async_add_entities)
+
+
+class DemoImage(ImageEntity):
+    """Representation of an image entity."""
+
+    def __init__(
+        self,
+        unique_id: str,
+        name: str,
+        content_type: str,
+        image: str,
+    ) -> None:
+        """Initialize the image entity."""
+        super().__init__()
+        self._attr_content_type = content_type
+        self._attr_name = name
+        self._attr_unique_id = unique_id
+        self._image_filename = image
+
+    async def async_added_to_hass(self):
+        """Set the update time."""
+        self._attr_image_last_updated = dt_util.utcnow()
+
+    async def async_image(self) -> bytes | None:
+        """Return bytes of image."""
+        image_path = Path(__file__).parent / self._image_filename
+        return await self.hass.async_add_executor_job(image_path.read_bytes)

homeassistant/const.py

@@ -37,6 +37,7 @@ class Platform(StrEnum):
     FAN = "fan"
     GEO_LOCATION = "geo_location"
     HUMIDIFIER = "humidifier"
+    IMAGE = "image"
     IMAGE_PROCESSING = "image_processing"
     LIGHT = "light"
     LOCK = "lock"

mypy.ini

@@ -1482,6 +1482,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.image.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.image_processing.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

tests/components/image/__init__.py

@@ -0,0 +1 @@
+"""The tests for the image integration."""

tests/components/image/conftest.py

@@ -0,0 +1,160 @@
+"""Test helpers for image."""
+from collections.abc import Generator
+
+import pytest
+
+from homeassistant.components import image
+from homeassistant.config_entries import ConfigEntry, ConfigFlow
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from homeassistant.setup import async_setup_component
+from homeassistant.util import dt as dt_util
+
+from tests.common import (
+    MockConfigEntry,
+    MockModule,
+    mock_config_flow,
+    mock_integration,
+    mock_platform,
+)
+
+TEST_DOMAIN = "test"
+
+
+class MockImageEntity(image.ImageEntity):
+    """Mock image entity."""
+
+    _attr_name = "Test"
+
+    async def async_added_to_hass(self):
+        """Set the update time."""
+        self._attr_image_last_updated = dt_util.utcnow()
+
+    async def async_image(self) -> bytes | None:
+        """Return bytes of image."""
+        return b"Test"
+
+
+class MockImageNoStateEntity(image.ImageEntity):
+    """Mock image entity."""
+
+    _attr_name = "Test"
+
+    async def async_image(self) -> bytes | None:
+        """Return bytes of image."""
+        return b"Test"
+
+
+class MockImageSyncEntity(image.ImageEntity):
+    """Mock image entity."""
+
+    _attr_name = "Test"
+
+    async def async_added_to_hass(self):
+        """Set the update time."""
+        self._attr_image_last_updated = dt_util.utcnow()
+
+    def image(self) -> bytes | None:
+        """Return bytes of image."""
+        return b"Test"
+
+
+class MockImageConfigEntry:
+    """A mock image config entry."""
+
+    def __init__(self, entities: list[image.ImageEntity]) -> None:
+        """Initialize."""
+        self._entities = entities
+
+    async def async_setup_entry(
+        self,
+        hass: HomeAssistant,
+        config_entry: ConfigEntry,
+        async_add_entities: AddEntitiesCallback,
+    ) -> None:
+        """Set up test image platform via config entry."""
+        async_add_entities([self._entities])
+
+
+class MockImagePlatform:
+    """A mock image platform."""
+
+    PLATFORM_SCHEMA = image.PLATFORM_SCHEMA
+
+    def __init__(self, entities: list[image.ImageEntity]) -> None:
+        """Initialize."""
+        self._entities = entities
+
+    async def async_setup_platform(
+        self,
+        hass: HomeAssistant,
+        config: ConfigType,
+        async_add_entities: AddEntitiesCallback,
+        discovery_info: DiscoveryInfoType | None = None,
+    ) -> None:
+        """Set up the mock image platform."""
+        async_add_entities(self._entities)
+
+
+@pytest.fixture(name="config_flow")
+def config_flow_fixture(hass: HomeAssistant) -> Generator[None, None, None]:
+    """Mock config flow."""
+
+    class MockFlow(ConfigFlow):
+        """Test flow."""
+
+    mock_platform(hass, f"{TEST_DOMAIN}.config_flow")
+
+    with mock_config_flow(TEST_DOMAIN, MockFlow):
+        yield
+
+
+@pytest.fixture(name="mock_image_config_entry")
+async def mock_image_config_entry_fixture(hass: HomeAssistant, config_flow):
+    """Initialize a mock image config_entry."""
+
+    async def async_setup_entry_init(
+        hass: HomeAssistant, config_entry: ConfigEntry
+    ) -> bool:
+        """Set up test config entry."""
+        await hass.config_entries.async_forward_entry_setup(config_entry, image.DOMAIN)
+        return True
+
+    async def async_unload_entry_init(
+        hass: HomeAssistant, config_entry: ConfigEntry
+    ) -> bool:
+        """Unload test config entry."""
+        await hass.config_entries.async_forward_entry_unload(config_entry, image.DOMAIN)
+        return True
+
+    mock_integration(
+        hass,
+        MockModule(
+            TEST_DOMAIN,
+            async_setup_entry=async_setup_entry_init,
+            async_unload_entry=async_unload_entry_init,
+        ),
+    )
+
+    mock_platform(
+        hass, f"{TEST_DOMAIN}.{image.DOMAIN}", MockImageConfigEntry(MockImageEntity())
+    )
+
+    config_entry = MockConfigEntry(domain=TEST_DOMAIN)
+    config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    return config_entry
+
+
+@pytest.fixture(name="mock_image_platform")
+async def mock_image_platform_fixture(hass: HomeAssistant):
+    """Initialize a mock image platform."""
+    mock_integration(hass, MockModule(domain="test"))
+    mock_platform(hass, "test.image", MockImagePlatform([MockImageEntity()]))
+    assert await async_setup_component(
+        hass, image.DOMAIN, {"image": {"platform": "test"}}
+    )
+    await hass.async_block_till_done()

tests/components/image/test_init.py

@@ -0,0 +1,169 @@
+"""The tests for the image component."""
+from http import HTTPStatus
+from unittest.mock import patch
+
+from aiohttp import hdrs
+import pytest
+
+from homeassistant.components import image
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from .conftest import (
+    MockImageEntity,
+    MockImageNoStateEntity,
+    MockImagePlatform,
+    MockImageSyncEntity,
+)
+
+from tests.common import MockModule, mock_integration, mock_platform
+from tests.typing import ClientSessionGenerator
+
+
+@pytest.mark.freeze_time("2023-04-01 00:00:00+00:00")
+async def test_state(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_image_platform
+) -> None:
+    """Test image state."""
+    state = hass.states.get("image.test")
+    assert state.state == "2023-04-01T00:00:00+00:00"
+    access_token = state.attributes["access_token"]
+    assert state.attributes == {
+        "access_token": access_token,
+        "entity_picture": f"/api/image_proxy/image.test?token={access_token}",
+        "friendly_name": "Test",
+    }
+
+
+@pytest.mark.freeze_time("2023-04-01 00:00:00+00:00")
+async def test_config_entry(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_image_config_entry
+) -> None:
+    """Test setting up an image platform from a config entry."""
+    state = hass.states.get("image.test")
+    assert state.state == "2023-04-01T00:00:00+00:00"
+    access_token = state.attributes["access_token"]
+    assert state.attributes == {
+        "access_token": access_token,
+        "entity_picture": f"/api/image_proxy/image.test?token={access_token}",
+        "friendly_name": "Test",
+    }
+
+
+@pytest.mark.freeze_time("2023-04-01 00:00:00+00:00")
+async def test_state_attr(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator
+) -> None:
+    """Test image state with entity picture from attr."""
+    mock_integration(hass, MockModule(domain="test"))
+    entity = MockImageEntity()
+    entity._attr_entity_picture = "abcd"
+    mock_platform(hass, "test.image", MockImagePlatform([entity]))
+    assert await async_setup_component(
+        hass, image.DOMAIN, {"image": {"platform": "test"}}
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get("image.test")
+    assert state.state == "2023-04-01T00:00:00+00:00"
+    access_token = state.attributes["access_token"]
+    assert state.attributes == {
+        "access_token": access_token,
+        "entity_picture": "abcd",
+        "friendly_name": "Test",
+    }
+
+
+async def test_no_state(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator
+) -> None:
+    """Test image state."""
+    mock_integration(hass, MockModule(domain="test"))
+    mock_platform(hass, "test.image", MockImagePlatform([MockImageNoStateEntity()]))
+    assert await async_setup_component(
+        hass, image.DOMAIN, {"image": {"platform": "test"}}
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get("image.test")
+    assert state.state == "unknown"
+    access_token = state.attributes["access_token"]
+    assert state.attributes == {
+        "access_token": access_token,
+        "entity_picture": f"/api/image_proxy/image.test?token={access_token}",
+        "friendly_name": "Test",
+    }
+
+
+async def test_fetch_image_authenticated(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_image_platform
+) -> None:
+    """Test fetching an image with an authenticated client."""
+    client = await hass_client()
+
+    resp = await client.get("/api/image_proxy/image.test")
+    assert resp.status == HTTPStatus.OK
+    body = await resp.read()
+    assert body == b"Test"
+
+    resp = await client.get("/api/image_proxy/image.unknown")
+    assert resp.status == HTTPStatus.NOT_FOUND
+
+
+async def test_fetch_image_fail(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator, mock_image_platform
+) -> None:
+    """Test fetching an image with an authenticated client."""
+    client = await hass_client()
+
+    with patch.object(MockImageEntity, "async_image", side_effect=TimeoutError):
+        resp = await client.get("/api/image_proxy/image.test")
+        assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
+
+
+async def test_fetch_image_sync(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator
+) -> None:
+    """Test fetching an image with an authenticated client."""
+    mock_integration(hass, MockModule(domain="test"))
+    mock_platform(hass, "test.image", MockImagePlatform([MockImageSyncEntity()]))
+    assert await async_setup_component(
+        hass, image.DOMAIN, {"image": {"platform": "test"}}
+    )
+    await hass.async_block_till_done()
+
+    client = await hass_client()
+
+    resp = await client.get("/api/image_proxy/image.test")
+    assert resp.status == HTTPStatus.OK
+    body = await resp.read()
+    assert body == b"Test"
+
+
+async def test_fetch_image_unauthenticated(
+    hass: HomeAssistant,
+    hass_client_no_auth: ClientSessionGenerator,
+    mock_image_platform,
+) -> None:
+    """Test fetching an image with an unauthenticated client."""
+    client = await hass_client_no_auth()
+
+    resp = await client.get("/api/image_proxy/image.test")
+    assert resp.status == HTTPStatus.FORBIDDEN
+
+    resp = await client.get("/api/image_proxy/image.test")
+    assert resp.status == HTTPStatus.FORBIDDEN
+
+    resp = await client.get(
+        "/api/image_proxy/image.test", headers={hdrs.AUTHORIZATION: "blabla"}
+    )
+    assert resp.status == HTTPStatus.UNAUTHORIZED
+
+    state = hass.states.get("image.test")
+    resp = await client.get(state.attributes["entity_picture"])
+    assert resp.status == HTTPStatus.OK
+    body = await resp.read()
+    assert body == b"Test"
+
+    resp = await client.get("/api/image_proxy/image.unknown")
+    assert resp.status == HTTPStatus.NOT_FOUND

tests/components/image/test_recorder.py

@@ -0,0 +1,40 @@
+"""The tests for image recorder."""
+from __future__ import annotations
+
+from datetime import timedelta
+
+from homeassistant.components.recorder import Recorder
+from homeassistant.components.recorder.history import get_significant_states
+from homeassistant.const import (
+    ATTR_ATTRIBUTION,
+    ATTR_ENTITY_PICTURE,
+    ATTR_FRIENDLY_NAME,
+    ATTR_SUPPORTED_FEATURES,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt as dt_util
+
+from tests.common import async_fire_time_changed
+from tests.components.recorder.common import async_wait_recording_done
+
+
+async def test_exclude_attributes(
+    recorder_mock: Recorder, hass: HomeAssistant, mock_image_platform
+) -> None:
+    """Test camera registered attributes to be excluded."""
+    now = dt_util.utcnow()
+    async_fire_time_changed(hass, dt_util.utcnow() + timedelta(minutes=5))
+    await hass.async_block_till_done()
+    await async_wait_recording_done(hass)
+
+    states = await hass.async_add_executor_job(
+        get_significant_states, hass, now, None, hass.states.async_entity_ids()
+    )
+    assert len(states) == 1
+    for entity_states in states.values():
+        for state in entity_states:
+            assert "access_token" not in state.attributes
+            assert ATTR_ENTITY_PICTURE not in state.attributes
+            assert ATTR_ATTRIBUTION not in state.attributes
+            assert ATTR_SUPPORTED_FEATURES not in state.attributes
+            assert ATTR_FRIENDLY_NAME in state.attributes

tests/components/kitchen_sink/test_image.py

@@ -0,0 +1,60 @@
+"""The tests for the kitchen_sink image platform."""
+from http import HTTPStatus
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.kitchen_sink import DOMAIN, image
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from tests.typing import ClientSessionGenerator
+
+
+@pytest.fixture
+async def image_only() -> None:
+    """Enable only the image platform."""
+    with patch(
+        "homeassistant.components.kitchen_sink.COMPONENTS_WITH_DEMO_PLATFORM",
+        [Platform.IMAGE],
+    ):
+        yield
+
+
+@pytest.fixture(autouse=True)
+async def setup_comp(hass: HomeAssistant, image_only):
+    """Set up demo component."""
+    assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
+    await hass.async_block_till_done()
+
+
+async def test_states(hass: HomeAssistant) -> None:
+    """Test the expected image entities are added."""
+    states = hass.states.async_all()
+    assert len(states) == 1
+    state = states[0]
+
+    access_token = state.attributes["access_token"]
+    assert state.entity_id == "image.qr_code"
+    assert state.attributes == {
+        "access_token": access_token,
+        "entity_picture": f"/api/image_proxy/image.qr_code?token={access_token}",
+        "friendly_name": "QR Code",
+    }
+
+
+async def test_fetch_image(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator
+) -> None:
+    """Test fetching an image with an authenticated client."""
+    client = await hass_client()
+
+    image_path = Path(image.__file__).parent / "qr_code.png"
+    expected_data = await hass.async_add_executor_job(image_path.read_bytes)
+
+    resp = await client.get("/api/image_proxy/image.qr_code")
+    assert resp.status == HTTPStatus.OK
+    body = await resp.read()
+    assert body == expected_data