mirror of
https://github.com/home-assistant/core.git
synced 2026-10-06 06:15:47 -04:00
Add diagnostics support for Collection Image integration (#183956)
This commit is contained in:
karwosts
@@ -0,0 +1,102 @@
|
||||
"""Diagnostics support for Collection Image."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant.components.image import DOMAIN as IMAGE_DOMAIN
|
||||
from homeassistant.components.media_player import BrowseError, MediaClass
|
||||
from homeassistant.components.media_source import (
|
||||
Unresolvable,
|
||||
async_browse_media,
|
||||
async_resolve_media,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .const import CONF_MEDIA, DOMAIN
|
||||
|
||||
|
||||
async def _async_browse_source(
|
||||
hass: HomeAssistant, media_content_id: str
|
||||
) -> dict[str, Any]:
|
||||
"""Browse a configured media source and describe its contents."""
|
||||
try:
|
||||
media = await async_browse_media(hass, media_content_id)
|
||||
except BrowseError as err:
|
||||
return {"media_content_id": media_content_id, "error": str(err)}
|
||||
|
||||
children = media.children or []
|
||||
return {
|
||||
"media_content_id": media_content_id,
|
||||
"title": media.title,
|
||||
"media_class": media.media_class,
|
||||
"children_count": len(children),
|
||||
"image_count": sum(
|
||||
1 for child in children if child.media_class == MediaClass.IMAGE
|
||||
),
|
||||
"children": [
|
||||
{
|
||||
"title": child.title,
|
||||
"media_content_id": child.media_content_id,
|
||||
"media_content_type": child.media_content_type,
|
||||
"media_class": child.media_class,
|
||||
"can_play": child.can_play,
|
||||
"can_expand": child.can_expand,
|
||||
}
|
||||
for child in children
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
async def _async_resolve_image(
|
||||
hass: HomeAssistant, media_content_id: str, entity_id: str
|
||||
) -> dict[str, Any]:
|
||||
"""Resolve the current image and describe the result."""
|
||||
try:
|
||||
resolved = await async_resolve_media(hass, media_content_id, entity_id)
|
||||
except Unresolvable as err:
|
||||
return {"error": str(err)}
|
||||
|
||||
return {
|
||||
# Some media sources put access tokens in the query string.
|
||||
"url": str(URL(resolved.url).with_query(None)) if resolved.url else None,
|
||||
"mime_type": resolved.mime_type,
|
||||
"path": str(resolved.path) if resolved.path else None,
|
||||
}
|
||||
|
||||
|
||||
async def async_get_config_entry_diagnostics(
|
||||
hass: HomeAssistant, entry: ConfigEntry
|
||||
) -> dict[str, Any]:
|
||||
"""Return diagnostics for a config entry."""
|
||||
media = entry.data[CONF_MEDIA]
|
||||
items = [media] if isinstance(media, dict) else media
|
||||
|
||||
current_image: dict[str, Any] | None = None
|
||||
entity_registry = er.async_get(hass)
|
||||
if (
|
||||
entity_id := entity_registry.async_get_entity_id(
|
||||
IMAGE_DOMAIN, DOMAIN, entry.entry_id
|
||||
)
|
||||
) and (state := hass.states.get(entity_id)):
|
||||
media_content_id = state.attributes.get("current_media_id")
|
||||
current_image = {
|
||||
"entity_id": entity_id,
|
||||
"state": state.state,
|
||||
"media_content_id": media_content_id,
|
||||
"resolved": (
|
||||
await _async_resolve_image(hass, media_content_id, entity_id)
|
||||
if media_content_id
|
||||
else None
|
||||
),
|
||||
}
|
||||
|
||||
return {
|
||||
"entry": {"title": entry.title, "data": dict(entry.data)},
|
||||
"current_image": current_image,
|
||||
"media_sources": [
|
||||
await _async_browse_source(hass, item["media_content_id"]) for item in items
|
||||
],
|
||||
}
|
||||
@@ -63,7 +63,7 @@ rules:
|
||||
devices:
|
||||
status: exempt
|
||||
comment: Adding a device adds no benefit for a single entity config entry.
|
||||
diagnostics: todo
|
||||
diagnostics: done
|
||||
discovery-update-info:
|
||||
status: exempt
|
||||
comment: Nothing to discover.
|
||||
|
||||
@@ -43,7 +43,9 @@ class MediaSourceMocks:
|
||||
|
||||
config_flow_browse: AsyncMock
|
||||
image_browse: AsyncMock
|
||||
diagnostics_browse: AsyncMock
|
||||
resolve: AsyncMock
|
||||
diagnostics_resolve: AsyncMock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -151,13 +153,23 @@ def mock_media_source(
|
||||
"homeassistant.components.collection_image.image.async_browse_media",
|
||||
new=AsyncMock(side_effect=browse_side_effect),
|
||||
) as image_browse,
|
||||
patch(
|
||||
"homeassistant.components.collection_image.diagnostics.async_browse_media",
|
||||
new=AsyncMock(side_effect=browse_side_effect),
|
||||
) as diagnostics_browse,
|
||||
patch(
|
||||
"homeassistant.components.collection_image.image.async_resolve_media",
|
||||
new=AsyncMock(side_effect=resolve_side_effect),
|
||||
) as resolve,
|
||||
patch(
|
||||
"homeassistant.components.collection_image.diagnostics.async_resolve_media",
|
||||
new=AsyncMock(side_effect=resolve_side_effect),
|
||||
) as diagnostics_resolve,
|
||||
):
|
||||
yield MediaSourceMocks(
|
||||
config_flow_browse=config_flow_browse,
|
||||
image_browse=image_browse,
|
||||
diagnostics_browse=diagnostics_browse,
|
||||
resolve=resolve,
|
||||
diagnostics_resolve=diagnostics_resolve,
|
||||
)
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
# serializer version: 1
|
||||
# name: test_diagnostics[path]
|
||||
dict({
|
||||
'current_image': dict({
|
||||
'entity_id': 'image.random_image',
|
||||
'media_content_id': 'media-source://mymedia/photo',
|
||||
'resolved': dict({
|
||||
'mime_type': 'image/png',
|
||||
'path': '/media/photo.png',
|
||||
'url': None,
|
||||
}),
|
||||
'state': '2025-11-08T12:00:00+00:00',
|
||||
}),
|
||||
'entry': dict({
|
||||
'data': dict({
|
||||
'media': list([
|
||||
dict({
|
||||
'media_content_id': 'media-source://mymedia',
|
||||
'media_content_type': '',
|
||||
'metadata': dict({
|
||||
'a': 'b',
|
||||
}),
|
||||
}),
|
||||
dict({
|
||||
'media_content_id': 'media-source://mymedia_empty',
|
||||
'media_content_type': '',
|
||||
'metadata': dict({
|
||||
'a': 'b',
|
||||
}),
|
||||
}),
|
||||
dict({
|
||||
'media_content_id': 'media-source://mymedia_error',
|
||||
'media_content_type': '',
|
||||
'metadata': dict({
|
||||
'a': 'b',
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
'title': 'Random Image',
|
||||
}),
|
||||
'media_sources': list([
|
||||
dict({
|
||||
'children': list([
|
||||
dict({
|
||||
'can_expand': False,
|
||||
'can_play': True,
|
||||
'media_class': 'music',
|
||||
'media_content_id': 'media-source://mymedia/music',
|
||||
'media_content_type': 'audio/mp3',
|
||||
'title': 'a music track',
|
||||
}),
|
||||
dict({
|
||||
'can_expand': False,
|
||||
'can_play': True,
|
||||
'media_class': 'image',
|
||||
'media_content_id': 'media-source://mymedia/photo',
|
||||
'media_content_type': 'image/png',
|
||||
'title': 'a picture',
|
||||
}),
|
||||
]),
|
||||
'children_count': 2,
|
||||
'image_count': 1,
|
||||
'media_class': '',
|
||||
'media_content_id': 'media-source://mymedia',
|
||||
'title': 'My pictures',
|
||||
}),
|
||||
dict({
|
||||
'children': list([
|
||||
]),
|
||||
'children_count': 0,
|
||||
'image_count': 0,
|
||||
'media_class': '',
|
||||
'media_content_id': 'media-source://mymedia_empty',
|
||||
'title': 'Empty folder',
|
||||
}),
|
||||
dict({
|
||||
'error': 'Mock directory failed to browse',
|
||||
'media_content_id': 'media-source://mymedia_error',
|
||||
}),
|
||||
]),
|
||||
})
|
||||
# ---
|
||||
# name: test_diagnostics[unavailable]
|
||||
dict({
|
||||
'current_image': dict({
|
||||
'entity_id': 'image.random_image',
|
||||
'media_content_id': None,
|
||||
'resolved': None,
|
||||
'state': 'unavailable',
|
||||
}),
|
||||
'entry': dict({
|
||||
'data': dict({
|
||||
'media': list([
|
||||
dict({
|
||||
'media_content_id': 'media-source://mymedia_empty',
|
||||
'media_content_type': '',
|
||||
'metadata': dict({
|
||||
'a': 'b',
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
'title': 'Random Image',
|
||||
}),
|
||||
'media_sources': list([
|
||||
dict({
|
||||
'children': list([
|
||||
]),
|
||||
'children_count': 0,
|
||||
'image_count': 0,
|
||||
'media_class': '',
|
||||
'media_content_id': 'media-source://mymedia_empty',
|
||||
'title': 'Empty folder',
|
||||
}),
|
||||
]),
|
||||
})
|
||||
# ---
|
||||
# name: test_diagnostics[url]
|
||||
dict({
|
||||
'current_image': dict({
|
||||
'entity_id': 'image.random_image',
|
||||
'media_content_id': 'media-source://mymedia/photo',
|
||||
'resolved': dict({
|
||||
'mime_type': 'image/png',
|
||||
'path': None,
|
||||
'url': 'https://example.com/photo.png',
|
||||
}),
|
||||
'state': '2025-11-08T12:00:00+00:00',
|
||||
}),
|
||||
'entry': dict({
|
||||
'data': dict({
|
||||
'media': list([
|
||||
dict({
|
||||
'media_content_id': 'media-source://mymedia',
|
||||
'media_content_type': '',
|
||||
'metadata': dict({
|
||||
'a': 'b',
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
}),
|
||||
'title': 'Random Image',
|
||||
}),
|
||||
'media_sources': list([
|
||||
dict({
|
||||
'children': list([
|
||||
dict({
|
||||
'can_expand': False,
|
||||
'can_play': True,
|
||||
'media_class': 'music',
|
||||
'media_content_id': 'media-source://mymedia/music',
|
||||
'media_content_type': 'audio/mp3',
|
||||
'title': 'a music track',
|
||||
}),
|
||||
dict({
|
||||
'can_expand': False,
|
||||
'can_play': True,
|
||||
'media_class': 'image',
|
||||
'media_content_id': 'media-source://mymedia/photo',
|
||||
'media_content_type': 'image/png',
|
||||
'title': 'a picture',
|
||||
}),
|
||||
]),
|
||||
'children_count': 2,
|
||||
'image_count': 1,
|
||||
'media_class': '',
|
||||
'media_content_id': 'media-source://mymedia',
|
||||
'title': 'My pictures',
|
||||
}),
|
||||
]),
|
||||
})
|
||||
# ---
|
||||
@@ -0,0 +1,124 @@
|
||||
"""Tests for the Collection Image diagnostics."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from freezegun import freeze_time
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.collection_image.const import DOMAIN
|
||||
from homeassistant.components.image import DOMAIN as IMAGE_DOMAIN
|
||||
from homeassistant.components.media_source import PlayMedia, Unresolvable
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from .conftest import MediaSourceState
|
||||
from .const import (
|
||||
MOCK_MEDIA_DIR_URI_1,
|
||||
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
|
||||
MOCK_MEDIA_DIR_URI_EMPTY,
|
||||
MOCK_MEDIA_IMAGE_URI_1,
|
||||
)
|
||||
from .helpers import config_entry_from_uri
|
||||
|
||||
from tests.components.diagnostics import get_diagnostics_for_config_entry
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_media_source")
|
||||
@pytest.mark.parametrize(
|
||||
("uris", "play_media"),
|
||||
[
|
||||
pytest.param(
|
||||
[
|
||||
MOCK_MEDIA_DIR_URI_1,
|
||||
MOCK_MEDIA_DIR_URI_EMPTY,
|
||||
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
|
||||
],
|
||||
PlayMedia(url="", mime_type="image/png", path=Path("/media/photo.png")),
|
||||
id="path",
|
||||
),
|
||||
pytest.param(
|
||||
[MOCK_MEDIA_DIR_URI_1],
|
||||
PlayMedia(
|
||||
url="https://example.com/photo.png?token=secret",
|
||||
mime_type="image/png",
|
||||
),
|
||||
id="url",
|
||||
),
|
||||
pytest.param(
|
||||
[MOCK_MEDIA_DIR_URI_EMPTY],
|
||||
PlayMedia(url="", mime_type="image/png", path=Path("/media/photo.png")),
|
||||
id="unavailable",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_diagnostics(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
snapshot: SnapshotAssertion,
|
||||
media_source_state: MediaSourceState,
|
||||
uris: list[str],
|
||||
play_media: PlayMedia,
|
||||
) -> None:
|
||||
"""Test config entry diagnostics."""
|
||||
media_source_state.resolve_results[MOCK_MEDIA_IMAGE_URI_1] = play_media
|
||||
config_entry = config_entry_from_uri(uris)
|
||||
config_entry.add_to_hass(hass)
|
||||
with freeze_time("2025-11-08T12:00:00+00:00"):
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert (
|
||||
await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
|
||||
== snapshot
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_media_source")
|
||||
async def test_diagnostics_entity_disabled(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test diagnostics when the image entity has no state."""
|
||||
config_entry = config_entry_from_uri(MOCK_MEDIA_DIR_URI_1)
|
||||
config_entry.add_to_hass(hass)
|
||||
entity_registry.async_get_or_create(
|
||||
IMAGE_DOMAIN,
|
||||
DOMAIN,
|
||||
config_entry.entry_id,
|
||||
config_entry=config_entry,
|
||||
disabled_by=er.RegistryEntryDisabler.USER,
|
||||
)
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
diagnostics = await get_diagnostics_for_config_entry(
|
||||
hass, hass_client, config_entry
|
||||
)
|
||||
|
||||
assert diagnostics["current_image"] is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_media_source")
|
||||
async def test_diagnostics_unresolvable(
|
||||
hass: HomeAssistant,
|
||||
hass_client: ClientSessionGenerator,
|
||||
media_source_state: MediaSourceState,
|
||||
) -> None:
|
||||
"""Test diagnostics when the current image can no longer be resolved."""
|
||||
config_entry = config_entry_from_uri(MOCK_MEDIA_DIR_URI_1)
|
||||
config_entry.add_to_hass(hass)
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
media_source_state.resolve_exceptions[MOCK_MEDIA_IMAGE_URI_1] = Unresolvable(
|
||||
"Image was deleted"
|
||||
)
|
||||
|
||||
diagnostics = await get_diagnostics_for_config_entry(
|
||||
hass, hass_client, config_entry
|
||||
)
|
||||
|
||||
assert diagnostics["current_image"]["resolved"] == {"error": "Image was deleted"}
|
||||
Reference in New Issue
Block a user