Persist nest media events to disk backed storage (#61641)

* Persist nest media events to disk backed storage

Persist nest events in the media player to disk, targeting about ~500mb
per camera device as a cap. Events are stored in config/nest/event_media/.

Add a NestEventMediaStore is used for persistence. It has three main jobs:
- Read/write the key/value data that holds event data (event type, time, device, etc)
- Read/write media contents to disk
- Pick the filename for the media event based on device and event deatils

The nest event media manager library handles cache management and eviction, and by
default uses an in memory cache. Home Assistant nest integration now provides the
disk backed implementation, which is invoked by the nest library.

The store reads the event metadata key/value dict on startup, and then writes it
back with a short delay of 5 seconds to avoid unnecessary writes.

Future work planned includes:
- Possibly a small memory buffer for media objects themselves. This could make sense
  when adding thumbnails to the media player grid to avoid unnecessary fetches
- Transcoding mp4 clips to animated image previews

* Address style errors

* Cleanup from CI test/pylint/etc.

* Put media for each device into its own directory

* Update comments for media store

* Decrease # of events to lower disk requirements

Target more like 1k events, to reduce disk needs.

* Address PR feedback

* Update homeassistant/components/nest/media_source.py

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

* Ignore incorrect mypy in nest library

* Fix pylint errors

Co-authored-by: Paulus Schoutsen <paulus@home-assistant.io>

homeassistant/components/nest/__init__.py

@@ -51,7 +51,7 @@ from .const import (
 )
 from .events import EVENT_NAME_MAP, NEST_EVENT
 from .legacy import async_setup_legacy, async_setup_legacy_entry
-from .media_source import get_media_source_devices
+from .media_source import async_get_media_event_store, get_media_source_devices
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -87,10 +87,11 @@ PLATFORMS = [Platform.SENSOR, Platform.CAMERA, Platform.CLIMATE]
 WEB_AUTH_DOMAIN = DOMAIN
 INSTALLED_AUTH_DOMAIN = f"{DOMAIN}.installed"
 
-# Fetch media for events with an in memory cache. The largest media items
-# are mp4 clips at ~90kb each, so this totals a few MB per camera.
-# Note: Media for events can only be published within 30 seconds of the event
-EVENT_MEDIA_CACHE_SIZE = 64
+# Fetch media events with a disk backed cache, with a limit for each camera
+# device. The largest media items are mp4 clips at ~120kb each, and we target
+# ~125MB of storage per camera to try to balance a reasonable user experience
+# for event history not not filling the disk.
+EVENT_MEDIA_CACHE_SIZE = 1024  # number of events
 
 
 class WebAuth(config_entry_oauth2_flow.LocalOAuth2Implementation):
@@ -169,6 +170,8 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
         ),
     )
 
+    hass.http.register_view(NestEventMediaView(hass))
+
     return True
 
 
@@ -215,6 +218,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     # Keep media for last N events in memory
     subscriber.cache_policy.event_cache_size = EVENT_MEDIA_CACHE_SIZE
     subscriber.cache_policy.fetch = True
+    # Use disk backed event media store
+    subscriber.cache_policy.store = await async_get_media_event_store(hass, subscriber)
 
     callback = SignalUpdateCallback(hass)
     subscriber.set_update_callback(callback.async_handle_event)
@@ -248,8 +253,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     hass.config_entries.async_setup_platforms(entry, PLATFORMS)
 
-    hass.http.register_view(NestEventMediaView(hass))
-
     return True
 
 

homeassistant/components/nest/media_source.py

@@ -21,10 +21,13 @@ from __future__ import annotations
 from collections.abc import Mapping
 from dataclasses import dataclass
 import logging
+import os
 
 from google_nest_sdm.camera_traits import CameraClipPreviewTrait, CameraEventImageTrait
 from google_nest_sdm.device import Device
 from google_nest_sdm.event import EventImageType, ImageEventBase
+from google_nest_sdm.event_media import EventMediaStore
+from google_nest_sdm.google_nest_subscriber import GoogleNestSubscriber
 
 from homeassistant.components.media_player.const import (
     MEDIA_CLASS_DIRECTORY,
@@ -42,12 +45,14 @@ from homeassistant.components.media_source.models import (
     PlayMedia,
 )
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+from homeassistant.helpers.storage import Store
 from homeassistant.helpers.template import DATE_STR_FORMAT
-from homeassistant.util import dt as dt_util
+from homeassistant.util import dt as dt_util, raise_if_invalid_filename
 
 from .const import DATA_SUBSCRIBER, DOMAIN
 from .device_info import NestDeviceInfo
-from .events import MEDIA_SOURCE_EVENT_TITLE_MAP
+from .events import EVENT_NAME_MAP, MEDIA_SOURCE_EVENT_TITLE_MAP
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -56,6 +61,175 @@ DEVICE_TITLE_FORMAT = "{device_name}: Recent Events"
 CLIP_TITLE_FORMAT = "{event_name} @ {event_time}"
 EVENT_MEDIA_API_URL_FORMAT = "/api/nest/event_media/{device_id}/{event_id}"
 
+STORAGE_KEY = "nest.event_media"
+STORAGE_VERSION = 1
+# Buffer writes every few minutes (plus guaranteed to be written at shutdown)
+STORAGE_SAVE_DELAY_SECONDS = 120
+# Path under config directory
+MEDIA_PATH = f"{DOMAIN}/event_media"
+
+# Size of small in-memory disk cache to avoid excessive disk reads
+DISK_READ_LRU_MAX_SIZE = 32
+
+
+async def async_get_media_event_store(
+    hass: HomeAssistant, subscriber: GoogleNestSubscriber
+) -> EventMediaStore:
+    """Create the disk backed EventMediaStore."""
+    media_path = hass.config.path(MEDIA_PATH)
+
+    def mkdir() -> None:
+        os.makedirs(media_path, exist_ok=True)
+
+    await hass.async_add_executor_job(mkdir)
+    store = Store(hass, STORAGE_VERSION, STORAGE_KEY, private=True)
+    return NestEventMediaStore(hass, subscriber, store, media_path)
+
+
+class NestEventMediaStore(EventMediaStore):
+    """Storage hook to locally persist nest media for events.
+
+    This interface is meant to provide two storage features:
+    - media storage of events (jpgs, mp4s)
+    - metadata about events (e.g. motion, person), filename of the media, etc.
+
+    The default implementation in nest is in memory, and this allows the data
+    to be backed by disk.
+
+    The nest event media manager internal to the subscriber manages the lifetime
+    of individual objects stored here (e.g. purging when going over storage
+    limits). This store manages the addition/deletion once instructed.
+    """
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        subscriber: GoogleNestSubscriber,
+        store: Store,
+        media_path: str,
+    ) -> None:
+        """Initialize NestEventMediaStore."""
+        self._hass = hass
+        self._subscriber = subscriber
+        self._store = store
+        self._media_path = media_path
+        self._data: dict | None = None
+        self._devices: Mapping[str, str] | None = {}
+
+    async def async_load(self) -> dict | None:
+        """Load data."""
+        if self._data is None:
+            self._devices = await self._get_devices()
+            data = await self._store.async_load()
+            if data is None:
+                _LOGGER.debug("Loaded empty event store")
+                self._data = {}
+            elif isinstance(data, dict):
+                _LOGGER.debug("Loaded event store with %d records", len(data))
+                self._data = data
+            else:
+                raise ValueError(
+                    "Unexpected data in storage version={}, key={}".format(
+                        STORAGE_VERSION, STORAGE_KEY
+                    )
+                )
+        return self._data
+
+    async def async_save(self, data: dict) -> None:  # type: ignore[override]
+        """Save data."""
+        self._data = data
+
+        def provide_data() -> dict:
+            return data
+
+        self._store.async_delay_save(provide_data, STORAGE_SAVE_DELAY_SECONDS)
+
+    def get_media_key(self, device_id: str, event: ImageEventBase) -> str:
+        """Return the filename to use for a new event."""
+        # Convert a nest device id to a home assistant device id
+        device_id_str = (
+            self._devices.get(device_id, f"{device_id}-unknown_device")
+            if self._devices
+            else "unknown_device"
+        )
+        event_id_str = event.event_session_id
+        try:
+            raise_if_invalid_filename(event_id_str)
+        except ValueError:
+            event_id_str = ""
+        time_str = str(int(event.timestamp.timestamp()))
+        event_type_str = EVENT_NAME_MAP.get(event.event_type, "event")
+        suffix = "jpg" if event.event_image_type == EventImageType.IMAGE else "mp4"
+        return f"{device_id_str}/{time_str}-{event_id_str}-{event_type_str}.{suffix}"
+
+    def get_media_filename(self, media_key: str) -> str:
+        """Return the filename in storage for a media key."""
+        return f"{self._media_path}/{media_key}"
+
+    async def async_load_media(self, media_key: str) -> bytes | None:
+        """Load media content."""
+        filename = self.get_media_filename(media_key)
+
+        def load_media(filename: str) -> bytes | None:
+            if not os.path.exists(filename):
+                return None
+            _LOGGER.debug("Reading event media from disk store: %s", filename)
+            with open(filename, "rb") as media:
+                return media.read()
+
+        try:
+            return await self._hass.async_add_executor_job(load_media, filename)
+        except OSError as err:
+            _LOGGER.error("Unable to read media file: %s %s", filename, err)
+            return None
+
+    async def async_save_media(self, media_key: str, content: bytes) -> None:
+        """Write media content."""
+        filename = self.get_media_filename(media_key)
+
+        def save_media(filename: str, content: bytes) -> None:
+            os.makedirs(os.path.dirname(filename), exist_ok=True)
+            if os.path.exists(filename):
+                _LOGGER.debug(
+                    "Event media already exists, not overwriting: %s", filename
+                )
+                return
+            _LOGGER.debug("Saving event media to disk store: %s", filename)
+            with open(filename, "wb") as media:
+                media.write(content)
+
+        try:
+            await self._hass.async_add_executor_job(save_media, filename, content)
+        except OSError as err:
+            _LOGGER.error("Unable to write media file: %s %s", filename, err)
+
+    async def async_remove_media(self, media_key: str) -> None:
+        """Remove media content."""
+        filename = self.get_media_filename(media_key)
+
+        def remove_media(filename: str) -> None:
+            if not os.path.exists(filename):
+                return None
+            _LOGGER.debug("Removing event media from disk store: %s", filename)
+            os.remove(filename)
+
+        try:
+            await self._hass.async_add_executor_job(remove_media, filename)
+        except OSError as err:
+            _LOGGER.error("Unable to remove media file: %s %s", filename, err)
+
+    async def _get_devices(self) -> Mapping[str, str]:
+        """Return a mapping of nest device id to home assistant device id."""
+        device_registry = dr.async_get(self._hass)
+        device_manager = await self._subscriber.async_get_device_manager()
+        devices = {}
+        for device in device_manager.devices.values():
+            if device_entry := device_registry.async_get_device(
+                {(DOMAIN, device.name)}
+            ):
+                devices[device.name] = device_entry.id
+        return devices
+
 
 async def async_get_media_source(hass: HomeAssistant) -> MediaSource:
     """Set up Nest media source."""
@@ -69,7 +243,7 @@ async def get_media_source_devices(hass: HomeAssistant) -> Mapping[str, Device]:
         return {}
     subscriber = hass.data[DOMAIN][DATA_SUBSCRIBER]
     device_manager = await subscriber.async_get_device_manager()
-    device_registry = await hass.helpers.device_registry.async_get_registry()
+    device_registry = dr.async_get(hass)
     devices = {}
     for device in device_manager.devices.values():
         if not (

tests/components/nest/test_media_source.py

@@ -6,6 +6,10 @@ as media in the media source.
 
 import datetime
 from http import HTTPStatus
+import shutil
+from typing import Generator
+from unittest.mock import Mock, patch
+import uuid
 
 import aiohttp
 from google_nest_sdm.device import Device
@@ -19,9 +23,15 @@ from homeassistant.components.media_source.error import Unresolvable
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.template import DATE_STR_FORMAT
+from homeassistant.setup import async_setup_component
 import homeassistant.util.dt as dt_util
 
-from .common import async_setup_sdm_platform
+from .common import (
+    CONFIG,
+    FakeSubscriber,
+    async_setup_sdm_platform,
+    create_config_entry,
+)
 
 DOMAIN = "nest"
 DEVICE_ID = "example/api/device/id"
@@ -49,6 +59,7 @@ BATTERY_CAMERA_TRAITS = {
     "sdm.devices.traits.CameraPerson": {},
     "sdm.devices.traits.CameraMotion": {},
 }
+
 PERSON_EVENT = "sdm.devices.events.CameraPerson.Person"
 MOTION_EVENT = "sdm.devices.events.CameraMotion.Motion"
 
@@ -63,6 +74,17 @@ IMAGE_BYTES_FROM_EVENT = b"test url image bytes"
 IMAGE_AUTHORIZATION_HEADERS = {"Authorization": "Basic g.0.eventToken"}
 
 
+@pytest.fixture(autouse=True)
+def cleanup_media_storage(hass):
+    """Test cleanup, remove any media storage persisted during the test."""
+    tmp_path = str(uuid.uuid4())
+    m = Mock(spec=float)
+    m.return_value = tmp_path
+    with patch("homeassistant.components.nest.media_source.MEDIA_PATH", new_callable=m):
+        yield
+        shutil.rmtree(hass.config.path(tmp_path), ignore_errors=True)
+
+
 async def async_setup_devices(hass, auth, device_type, traits={}, events=[]):
     """Set up the platform and prerequisites."""
     devices = {
@@ -115,6 +137,22 @@ def create_event_message(event_data, timestamp, device_id=None):
     )
 
 
+def create_battery_event_data(
+    event_type, event_session_id=EVENT_SESSION_ID, event_id="n:2"
+):
+    """Return event payload data for a battery camera event."""
+    return {
+        event_type: {
+            "eventSessionId": event_session_id,
+            "eventId": event_id,
+        },
+        "sdm.devices.events.CameraClipPreview.ClipPreview": {
+            "eventSessionId": event_session_id,
+            "previewUrl": "https://127.0.0.1/example",
+        },
+    }
+
+
 async def test_no_eligible_devices(hass, auth):
     """Test a media source with no eligible camera devices."""
     await async_setup_devices(
@@ -335,7 +373,7 @@ async def test_event_order(hass, auth):
     event_timestamp_string = event_timestamp2.strftime(DATE_STR_FORMAT)
     assert browse.children[0].title == f"Motion @ {event_timestamp_string}"
     assert not browse.children[0].can_expand
-    assert not browse.can_play
+    assert not browse.children[0].can_play
 
     # Person event is next
     assert browse.children[1].domain == DOMAIN
@@ -344,7 +382,7 @@ async def test_event_order(hass, auth):
     event_timestamp_string = event_timestamp1.strftime(DATE_STR_FORMAT)
     assert browse.children[1].title == f"Person @ {event_timestamp_string}"
     assert not browse.children[1].can_expand
-    assert not browse.can_play
+    assert not browse.children[1].can_play
 
 
 async def test_browse_invalid_device_id(hass, auth):
@@ -436,16 +474,6 @@ async def test_resolve_invalid_event_id(hass, auth):
 async def test_camera_event_clip_preview(hass, auth, hass_client):
     """Test an event for a battery camera video clip."""
     event_timestamp = dt_util.now()
-    event_data = {
-        "sdm.devices.events.CameraMotion.Motion": {
-            "eventSessionId": EVENT_SESSION_ID,
-            "eventId": "n:2",
-        },
-        "sdm.devices.events.CameraClipPreview.ClipPreview": {
-            "eventSessionId": EVENT_SESSION_ID,
-            "previewUrl": "https://127.0.0.1/example",
-        },
-    }
     await async_setup_devices(
         hass,
         auth,
@@ -453,7 +481,7 @@ async def test_camera_event_clip_preview(hass, auth, hass_client):
         BATTERY_CAMERA_TRAITS,
         events=[
             create_event_message(
-                event_data,
+                create_battery_event_data(MOTION_EVENT),
                 timestamp=event_timestamp,
             ),
         ],
@@ -692,3 +720,314 @@ async def test_multiple_devices(hass, auth, hass_client):
         hass, f"{const.URI_SCHEME}{DOMAIN}/{device2.id}"
     )
     assert len(browse.children) == 3
+
+
+@pytest.fixture
+def event_store() -> Generator[None, None, None]:
+    """Persist changes to event store immediately."""
+    m = Mock(spec=float)
+    m.return_value = 0
+    with patch(
+        "homeassistant.components.nest.media_source.STORAGE_SAVE_DELAY_SECONDS",
+        new_callable=m,
+    ):
+        yield
+
+
+async def test_media_store_persistence(hass, auth, hass_client, event_store):
+    """Test the disk backed media store persistence."""
+    nest_device = Device.MakeDevice(
+        {
+            "name": DEVICE_ID,
+            "type": CAMERA_DEVICE_TYPE,
+            "traits": BATTERY_CAMERA_TRAITS,
+        },
+        auth=auth,
+    )
+
+    subscriber = FakeSubscriber()
+    device_manager = await subscriber.async_get_device_manager()
+    device_manager.add_device(nest_device)
+    # Fetch media for events when published
+    subscriber.cache_policy.fetch = True
+
+    config_entry = create_config_entry(hass)
+
+    with patch(
+        "homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
+    ), patch("homeassistant.components.nest.PLATFORMS", [PLATFORM]), patch(
+        "homeassistant.components.nest.api.GoogleNestSubscriber",
+        return_value=subscriber,
+    ):
+        assert await async_setup_component(hass, DOMAIN, CONFIG)
+        await hass.async_block_till_done()
+
+    device_registry = dr.async_get(hass)
+    device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
+    assert device
+    assert device.name == DEVICE_NAME
+
+    auth.responses = [
+        aiohttp.web.Response(body=IMAGE_BYTES_FROM_EVENT),
+    ]
+    event_timestamp = dt_util.now()
+    await subscriber.async_receive_event(
+        create_event_message(
+            create_battery_event_data(MOTION_EVENT), timestamp=event_timestamp
+        )
+    )
+
+    # Browse to event
+    browse = await media_source.async_browse_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}"
+    )
+    assert len(browse.children) == 1
+    assert browse.children[0].domain == DOMAIN
+    assert browse.children[0].identifier == f"{device.id}/{EVENT_SESSION_ID}"
+    event_timestamp_string = event_timestamp.strftime(DATE_STR_FORMAT)
+    assert browse.children[0].title == f"Motion @ {event_timestamp_string}"
+    assert not browse.children[0].can_expand
+    assert browse.children[0].can_play
+
+    media = await media_source.async_resolve_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}/{EVENT_SESSION_ID}"
+    )
+    assert media.url == f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}"
+    assert media.mime_type == "video/mp4"
+
+    # Fetch event media
+    client = await hass_client()
+    response = await client.get(media.url)
+    assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+    contents = await response.read()
+    assert contents == IMAGE_BYTES_FROM_EVENT
+
+    # Ensure event media store persists to disk
+    await hass.async_block_till_done()
+
+    # Unload the integration.
+    assert config_entry.state == ConfigEntryState.LOADED
+    assert await hass.config_entries.async_unload(config_entry.entry_id)
+    await hass.async_block_till_done()
+    assert config_entry.state == ConfigEntryState.NOT_LOADED
+
+    # Now rebuild the entire integration and verify that all persisted storage
+    # can be re-loaded from disk.
+    subscriber = FakeSubscriber()
+    device_manager = await subscriber.async_get_device_manager()
+    device_manager.add_device(nest_device)
+    # Fetch media for events when published
+    subscriber.cache_policy.fetch = True
+
+    with patch(
+        "homeassistant.helpers.config_entry_oauth2_flow.async_get_config_entry_implementation"
+    ), patch("homeassistant.components.nest.PLATFORMS", [PLATFORM]), patch(
+        "homeassistant.components.nest.api.GoogleNestSubscriber",
+        return_value=subscriber,
+    ):
+        await hass.config_entries.async_reload(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    device_registry = dr.async_get(hass)
+    device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
+    assert device
+    assert device.name == DEVICE_NAME
+
+    # Verify event metadata exists
+    browse = await media_source.async_browse_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}"
+    )
+    assert len(browse.children) == 1
+    assert browse.children[0].domain == DOMAIN
+    assert browse.children[0].identifier == f"{device.id}/{EVENT_SESSION_ID}"
+    event_timestamp_string = event_timestamp.strftime(DATE_STR_FORMAT)
+    assert browse.children[0].title == f"Motion @ {event_timestamp_string}"
+    assert not browse.children[0].can_expand
+    assert browse.children[0].can_play
+
+    media = await media_source.async_resolve_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}/{EVENT_SESSION_ID}"
+    )
+    assert media.url == f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}"
+    assert media.mime_type == "video/mp4"
+
+    # Verify media exists
+    response = await client.get(media.url)
+    assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+    contents = await response.read()
+    assert contents == IMAGE_BYTES_FROM_EVENT
+
+
+async def test_media_store_filesystem_error(hass, auth, hass_client):
+    """Test a filesystem error read/writing event media."""
+    event_timestamp = dt_util.now()
+    await async_setup_devices(
+        hass,
+        auth,
+        CAMERA_DEVICE_TYPE,
+        BATTERY_CAMERA_TRAITS,
+        events=[
+            create_event_message(
+                create_battery_event_data(MOTION_EVENT),
+                timestamp=event_timestamp,
+            ),
+        ],
+    )
+
+    assert len(hass.states.async_all()) == 1
+    camera = hass.states.get("camera.front")
+    assert camera is not None
+
+    device_registry = dr.async_get(hass)
+    device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
+    assert device
+    assert device.name == DEVICE_NAME
+
+    auth.responses = [
+        aiohttp.web.Response(body=IMAGE_BYTES_FROM_EVENT),
+    ]
+
+    # The client fetches the media from the server, but has a failure when
+    # persisting the media to disk.
+    client = await hass_client()
+    with patch("homeassistant.components.nest.media_source.open", side_effect=OSError):
+        response = await client.get(
+            f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}"
+        )
+        assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+        contents = await response.read()
+        assert contents == IMAGE_BYTES_FROM_EVENT
+        await hass.async_block_till_done()
+
+    # Fetch the media again, and since the object does not exist in the cache it
+    # needs to be fetched again. The server returns an error to prove that it was
+    # not a cache read. A second attempt succeeds.
+    auth.responses = [
+        aiohttp.web.Response(status=HTTPStatus.INTERNAL_SERVER_ERROR),
+        aiohttp.web.Response(body=IMAGE_BYTES_FROM_EVENT),
+    ]
+    # First attempt, server fails when fetching
+    response = await client.get(f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}")
+    assert response.status == HTTPStatus.INTERNAL_SERVER_ERROR, (
+        "Response not matched: %s" % response
+    )
+
+    # Second attempt, server responds success
+    response = await client.get(f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}")
+    assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+    contents = await response.read()
+    assert contents == IMAGE_BYTES_FROM_EVENT
+
+    # Third attempt reads from the disk cache with no server fetch
+    response = await client.get(f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}")
+    assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+    contents = await response.read()
+    assert contents == IMAGE_BYTES_FROM_EVENT
+
+    auth.responses = [
+        aiohttp.web.Response(body=IMAGE_BYTES_FROM_EVENT),
+    ]
+    # Exercise a failure reading from the disk cache. Re-populate from server and write to disk ok
+    with patch("homeassistant.components.nest.media_source.open", side_effect=OSError):
+        response = await client.get(
+            f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}"
+        )
+        assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+        contents = await response.read()
+        assert contents == IMAGE_BYTES_FROM_EVENT
+        await hass.async_block_till_done()
+
+    response = await client.get(f"/api/nest/event_media/{device.id}/{EVENT_SESSION_ID}")
+    assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+    contents = await response.read()
+    assert contents == IMAGE_BYTES_FROM_EVENT
+
+
+async def test_camera_event_media_eviction(hass, auth, hass_client):
+    """Test media files getting evicted from the cache."""
+
+    # Set small cache size for testing eviction
+    m = Mock(spec=float)
+    m.return_value = 5
+    with patch("homeassistant.components.nest.EVENT_MEDIA_CACHE_SIZE", new_callable=m):
+        subscriber = await async_setup_devices(
+            hass,
+            auth,
+            CAMERA_DEVICE_TYPE,
+            BATTERY_CAMERA_TRAITS,
+        )
+
+    # Media fetched as soon as it is published
+    subscriber.cache_policy.fetch = True
+
+    device_registry = dr.async_get(hass)
+    device = device_registry.async_get_device({(DOMAIN, DEVICE_ID)})
+    assert device
+    assert device.name == DEVICE_NAME
+
+    # Browse to the device
+    browse = await media_source.async_browse_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}"
+    )
+    assert browse.domain == DOMAIN
+    assert browse.identifier == device.id
+    assert browse.title == "Front: Recent Events"
+    assert browse.can_expand
+
+    # No events published yet
+    assert len(browse.children) == 0
+
+    event_timestamp = dt_util.now()
+    for i in range(0, 7):
+        auth.responses = [aiohttp.web.Response(body=f"image-bytes-{i}".encode())]
+        ts = event_timestamp + datetime.timedelta(seconds=i)
+        await subscriber.async_receive_event(
+            create_event_message(
+                create_battery_event_data(
+                    MOTION_EVENT, event_session_id=f"event-session-{i}"
+                ),
+                timestamp=ts,
+            )
+        )
+    await hass.async_block_till_done()
+
+    # Cache is limited to 5 events removing media as the cache is filled
+    browse = await media_source.async_browse_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}"
+    )
+    assert len(browse.children) == 5
+
+    auth.responses = [
+        aiohttp.web.Response(body=b"image-bytes-7"),
+    ]
+    ts = event_timestamp + datetime.timedelta(seconds=8)
+    # Simulate a failure case removing the media on cache eviction
+    with patch(
+        "homeassistant.components.nest.media_source.os.remove", side_effect=OSError
+    ) as mock_remove:
+        await subscriber.async_receive_event(
+            create_event_message(
+                create_battery_event_data(
+                    MOTION_EVENT, event_session_id="event-session-7"
+                ),
+                timestamp=ts,
+            )
+        )
+        await hass.async_block_till_done()
+        assert mock_remove.called
+
+    browse = await media_source.async_browse_media(
+        hass, f"{const.URI_SCHEME}{DOMAIN}/{device.id}"
+    )
+    assert len(browse.children) == 5
+
+    # Verify all other content is still persisted correctly
+    client = await hass_client()
+    for i in range(3, 8):
+        response = await client.get(
+            f"/api/nest/event_media/{device.id}/event-session-{i}"
+        )
+        assert response.status == HTTPStatus.OK, "Response not matched: %s" % response
+        contents = await response.read()
+        assert contents == f"image-bytes-{i}".encode()
+        await hass.async_block_till_done()