Refactor tests into more reusable fixtures (#181006)

This commit is contained in:
karwosts
2026-09-01 21:34:19 +02:00
committed by GitHub
parent e871b992d9
commit 263f394d9f
5 changed files with 357 additions and 263 deletions
+111 -44
View File
@@ -1,82 +1,149 @@
"""Fixtures for the Collection Image integration tests."""
from collections.abc import Iterator
from dataclasses import dataclass, field
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant.components.collection_image.const import DOMAIN
from homeassistant.components.media_player import BrowseMedia, MediaClass
from homeassistant.components.media_player import BrowseError, BrowseMedia, MediaClass
from homeassistant.components.media_source import BrowseMediaSource, PlayMedia
from homeassistant.core import HomeAssistant
from .const import TEST_IMAGE
from .const import (
MOCK_MEDIA_DIR_URI_1,
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
MOCK_MEDIA_DIR_URI_EMPTY,
MOCK_MEDIA_IMAGE_URI_1,
TEST_IMAGE,
)
from .helpers import directory, image
from tests.common import MockConfigEntry
@dataclass
class MediaSourceState:
"""Configurable responses for the mocked media-source API."""
browse_results: dict[str, BrowseMediaSource] = field(default_factory=dict)
browse_exceptions: dict[str, Exception] = field(default_factory=dict)
resolve_results: dict[str, PlayMedia] = field(default_factory=dict)
resolve_exceptions: dict[str, Exception] = field(default_factory=dict)
@dataclass(frozen=True)
class MediaSourceMocks:
"""Mocks installed for calls to the media-source API."""
config_flow_browse: AsyncMock
image_browse: AsyncMock
resolve: AsyncMock
@pytest.fixture
def config_entry() -> MockConfigEntry:
"""Return the default collection-image config entry."""
return MockConfigEntry(
domain=DOMAIN,
title="Random Image",
data={
"media": {
"media_content_id": "media-source://mymedia",
"media_content_id": MOCK_MEDIA_DIR_URI_1,
"media_content_type": "",
},
},
domain=DOMAIN,
title="Random Image",
)
@pytest.fixture
def browse_media_result() -> BrowseMediaSource:
"""Return a default collection containing one image."""
return BrowseMediaSource(
domain=None,
identifier=None,
media_class="",
media_content_type="",
title="",
can_play=False,
can_expand=True,
children=[
BrowseMedia(
media_class=MediaClass.MUSIC,
media_content_id="media-source://mymedia/music",
media_content_type="audio/mp3",
title="a music track",
can_play=True,
can_expand=False,
def media_source_state() -> MediaSourceState:
"""Return default configurable responses for the media-source mock."""
return MediaSourceState(
browse_results={
MOCK_MEDIA_DIR_URI_1: directory(
"My pictures",
BrowseMedia(
media_class=MediaClass.MUSIC,
media_content_id="media-source://mymedia/music",
media_content_type="audio/mp3",
title="a music track",
can_play=True,
can_expand=False,
),
image(MOCK_MEDIA_IMAGE_URI_1),
),
BrowseMedia(
media_class=MediaClass.IMAGE,
media_content_id="media-source://mymedia/photo",
media_content_type="image/png",
title="a picture",
can_play=True,
can_expand=False,
MOCK_MEDIA_DIR_URI_EMPTY: directory("Empty folder"),
},
browse_exceptions={
MOCK_MEDIA_DIR_URI_BROWSE_ERROR: BrowseError(
"Mock directory failed to browse"
)
},
resolve_results={
MOCK_MEDIA_IMAGE_URI_1: PlayMedia(
url="",
mime_type="image/png",
path=TEST_IMAGE,
),
],
},
)
@pytest.fixture
def mock_media_source(browse_media_result: BrowseMediaSource):
"""Mock browsing and resolving the configured media source."""
def mock_media_source(
media_source_state: MediaSourceState,
) -> Iterator[MediaSourceMocks]:
"""Patch media-source calls made by the collection-image integration."""
async def browse_side_effect(
_hass: HomeAssistant,
media_content_id: str,
*,
content_filter=None,
) -> BrowseMediaSource:
if exception := media_source_state.browse_exceptions.get(media_content_id):
raise exception
try:
return media_source_state.browse_results[media_content_id]
except KeyError as err:
raise ValueError(
f"Unexpected media content ID: {media_content_id}"
) from err
async def resolve_side_effect(
_hass: HomeAssistant,
media_content_id: str,
_entity_id: str,
) -> PlayMedia:
if exception := media_source_state.resolve_exceptions.get(media_content_id):
raise exception
try:
return media_source_state.resolve_results[media_content_id]
except KeyError as err:
raise ValueError(
f"Unexpected media content ID: {media_content_id}"
) from err
with (
patch(
"homeassistant.components.collection_image.config_flow.async_browse_media",
new=AsyncMock(side_effect=browse_side_effect),
) as config_flow_browse,
patch(
"homeassistant.components.collection_image.image.async_browse_media",
new=AsyncMock(return_value=browse_media_result),
) as mock_browse,
new=AsyncMock(side_effect=browse_side_effect),
) as image_browse,
patch(
"homeassistant.components.collection_image.image.async_resolve_media",
new=AsyncMock(
return_value=PlayMedia(
url="",
mime_type="image/png",
path=TEST_IMAGE,
)
),
) as mock_resolve,
new=AsyncMock(side_effect=resolve_side_effect),
) as resolve,
):
yield mock_browse, mock_resolve
yield MediaSourceMocks(
config_flow_browse=config_flow_browse,
image_browse=image_browse,
resolve=resolve,
)
@@ -4,3 +4,9 @@ from pathlib import Path
TEST_IMAGE = Path(__file__).parent / "test.png"
DEFAULT_ENTITY_ID = "image.random_image"
MOCK_MEDIA_DIR_URI_1 = "media-source://mymedia"
MOCK_MEDIA_DIR_URI_EMPTY = "media-source://mymedia_empty"
MOCK_MEDIA_DIR_URI_BROWSE_ERROR = "media-source://mymedia_error"
MOCK_MEDIA_IMAGE_URI_1 = "media-source://mymedia/photo"
@@ -0,0 +1,54 @@
"""Helper utilities for collection image tests."""
from homeassistant.components.collection_image.const import DOMAIN
from homeassistant.components.media_player import BrowseMedia, MediaClass
from homeassistant.components.media_source import BrowseMediaSource
from tests.common import MockConfigEntry
def config_entry_from_uri(uri: str) -> MockConfigEntry:
"""From a uri, construct a config entry."""
return MockConfigEntry(
data={
"media": {
"media_content_id": uri,
"media_content_type": "",
},
},
domain=DOMAIN,
title="Random Image",
)
def image(
media_content_id: str,
*,
title: str = "a picture",
) -> BrowseMedia:
"""Create a playable image browse result."""
return BrowseMedia(
media_class=MediaClass.IMAGE,
media_content_id=media_content_id,
media_content_type="image/png",
title=title,
can_play=True,
can_expand=False,
)
def directory(
title: str,
*children: BrowseMedia,
) -> BrowseMediaSource:
"""Create an expandable browse result."""
return BrowseMediaSource(
domain=None,
identifier=None,
media_class="",
media_content_type="",
title=title,
can_play=False,
can_expand=True,
children=list(children),
)
@@ -1,68 +1,43 @@
"""Test the Collection Image config flow."""
from unittest.mock import patch
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant import config_entries
from homeassistant.components.collection_image.const import DOMAIN
from homeassistant.components.media_player import BrowseMedia, MediaClass
from homeassistant.components.media_source import BrowseMediaSource
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .const import (
MOCK_MEDIA_DIR_URI_1,
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
MOCK_MEDIA_DIR_URI_EMPTY,
)
async def _assert_successful_configure(
hass: HomeAssistant, previous_step: config_entries.ConfigFlowResult
) -> None:
with (
patch(
"homeassistant.components.collection_image.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"homeassistant.components.collection_image.config_flow.async_browse_media",
return_value=BrowseMediaSource(
domain=None,
identifier=None,
media_class="",
media_content_type="",
title="My pictures",
can_play=False,
can_expand=True,
children=[
BrowseMedia(
media_class=MediaClass.IMAGE,
media_content_id="media-source://mymedia/photo",
media_content_type="image/png",
title="a picture",
can_play=True,
can_expand=False,
),
],
),
),
):
result = await hass.config_entries.flow.async_configure(
previous_step["flow_id"],
{
"media": {
"media_content_id": "media-source://mymedia",
"media_content_type": "",
},
},
)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == "My pictures collection"
assert result.get("data") == {
@pytest.fixture
def mock_setup_entry():
"""Mock collection_image setup successfully."""
with patch(
"homeassistant.components.collection_image.async_setup_entry",
new=AsyncMock(return_value=True),
) as mock_setup:
yield mock_setup
def _data_from_uri(uri: str) -> dict:
return {
"media": {
"media_content_id": "media-source://mymedia",
"media_content_id": uri,
"media_content_type": "",
},
}
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_config_flow(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_media_source")
async def test_config_flow(hass: HomeAssistant, mock_setup_entry) -> None:
"""Test the config flow."""
result = await hass.config_entries.flow.async_init(
@@ -71,11 +46,41 @@ async def test_config_flow(hass: HomeAssistant) -> None:
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {}
await _assert_successful_configure(hass, result)
data = _data_from_uri(MOCK_MEDIA_DIR_URI_1)
expected_title = "My pictures collection"
result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == expected_title
assert result.get("data") == data
assert len(mock_setup_entry.mock_calls) == 1
async def test_config_flow_with_error(hass: HomeAssistant) -> None:
"""Test the config flow with an invalid directory."""
@pytest.mark.parametrize(
("uri", "error", "placeholders"),
[
(
MOCK_MEDIA_DIR_URI_EMPTY,
"selected_media_no_images",
{},
),
(
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
"failed_browse",
{"error": "Mock directory failed to browse"},
),
],
)
@pytest.mark.usefixtures("mock_media_source")
async def test_config_flow_error(
hass: HomeAssistant,
mock_setup_entry,
uri: str,
error: str,
placeholders: dict,
) -> None:
"""Test the config flow with an invalid media."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -83,79 +88,24 @@ async def test_config_flow_with_error(hass: HomeAssistant) -> None:
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {}
with (
patch(
"homeassistant.components.collection_image.async_setup_entry",
return_value=True,
) as mock_setup_entry,
patch(
"homeassistant.components.collection_image.config_flow.async_browse_media",
return_value=BrowseMediaSource(
domain=None,
identifier=None,
media_class="",
media_content_type="",
title="",
can_play=False,
can_expand=True,
children=[],
),
),
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"media": {
"media_content_id": "media-source://mymedia_empty",
"media_content_type": "",
},
},
)
await hass.async_block_till_done()
data = _data_from_uri(uri)
result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.FORM
assert result.get("title") is None
assert result.get("data") is None
assert result.get("errors") == {"media": "selected_media_no_images"}
assert result.get("errors") == {"media": error}
assert result.get("description_placeholders") == placeholders
assert len(mock_setup_entry.mock_calls) == 0
# Try again successfully to ensure we can recover from errors
await _assert_successful_configure(hass, result)
data = _data_from_uri(MOCK_MEDIA_DIR_URI_1)
expected_title = "My pictures collection"
result = await hass.config_entries.flow.async_configure(result["flow_id"], data)
async def test_config_flow_with_exception(hass: HomeAssistant) -> None:
"""Test the config flow with a browse failure."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result.get("type") is FlowResultType.FORM
assert result.get("errors") == {}
with (
patch(
"homeassistant.components.collection_image.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
"media": {
"media_content_id": "media-source://mymedia",
"media_content_type": "",
},
},
)
await hass.async_block_till_done()
assert result.get("type") is FlowResultType.FORM
assert result.get("title") is None
assert result.get("data") is None
assert result.get("errors") == {"media": "failed_browse"}
assert result.get("description_placeholders") == {
"error": "Media Source not loaded"
}
assert len(mock_setup_entry.mock_calls) == 0
await _assert_successful_configure(hass, result)
assert result.get("type") is FlowResultType.CREATE_ENTRY
assert result.get("title") == expected_title
assert result.get("data") == data
assert len(mock_setup_entry.mock_calls) == 1
+116 -99
View File
@@ -7,9 +7,11 @@ from unittest.mock import AsyncMock, patch
from freezegun import freeze_time
import pytest
from homeassistant.components.collection_image import DOMAIN
from homeassistant.components.image import Image, async_get_image
from homeassistant.components.media_source import BrowseMediaSource, PlayMedia
from homeassistant.components.media_source import PlayMedia, Unresolvable
from homeassistant.const import (
ATTR_ENTITY_ID,
EVENT_HOMEASSISTANT_STARTED,
STATE_UNAVAILABLE,
STATE_UNKNOWN,
@@ -17,7 +19,15 @@ from homeassistant.const import (
from homeassistant.core import CoreState, HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .const import DEFAULT_ENTITY_ID, TEST_IMAGE
from .conftest import MediaSourceMocks, MediaSourceState
from .const import (
DEFAULT_ENTITY_ID,
MOCK_MEDIA_DIR_URI_BROWSE_ERROR,
MOCK_MEDIA_DIR_URI_EMPTY,
MOCK_MEDIA_IMAGE_URI_1,
TEST_IMAGE,
)
from .helpers import config_entry_from_uri
from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator
@@ -25,11 +35,25 @@ from tests.typing import ClientSessionGenerator
TEST_TIME = "2025-11-08T12:00:00+00:00"
async def _verify_path_image(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
):
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
assert resp.status == HTTPStatus.OK
assert resp.content_type == "image/png"
expected_data = await hass.async_add_executor_job(TEST_IMAGE.read_bytes)
body = await resp.read()
assert body == expected_data
@pytest.mark.usefixtures("mock_media_source")
async def test_image(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
mock_media_source,
) -> None:
"""Test loading an image."""
with (
@@ -43,21 +67,14 @@ async def test_image(
assert state and state.state == TEST_TIME
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
assert resp.status == HTTPStatus.OK
assert resp.content_type == "image/png"
expected_data = await hass.async_add_executor_job(TEST_IMAGE.read_bytes)
body = await resp.read()
assert body == expected_data
await _verify_path_image(hass, hass_client)
async def test_image_during_startup(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
mock_media_source,
mock_media_source: MediaSourceMocks,
) -> None:
"""Test loading an image, ensuring that we don't browse until after startup is complete."""
with freeze_time(TEST_TIME):
@@ -67,47 +84,39 @@ async def test_image_during_startup(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
mock_media_source.image_browse.assert_not_called()
mock_media_source.resolve.assert_not_called()
hass.set_state(CoreState.running)
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
await hass.async_block_till_done()
mock_media_source.image_browse.assert_awaited_once()
mock_media_source.resolve.assert_awaited_once()
state = hass.states.get(DEFAULT_ENTITY_ID)
assert state and state.state == TEST_TIME
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
assert resp.status == HTTPStatus.OK
assert resp.content_type == "image/png"
expected_data = await hass.async_add_executor_job(TEST_IMAGE.read_bytes)
body = await resp.read()
assert body == expected_data
await _verify_path_image(hass, hass_client)
@pytest.mark.usefixtures("mock_media_source")
async def test_image_url(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
browse_media_result: BrowseMediaSource,
media_source_state: MediaSourceState,
) -> None:
"""Test loading an image, when media resolves to a URL."""
media_source_state.resolve_results[MOCK_MEDIA_IMAGE_URI_1] = PlayMedia(
url="http://example.com/test.png",
mime_type="image/png",
)
expected_data = await hass.async_add_executor_job(TEST_IMAGE.read_bytes)
with (
freeze_time(TEST_TIME),
patch(
"homeassistant.components.collection_image.image.async_browse_media",
return_value=browse_media_result,
),
patch(
"homeassistant.components.collection_image.image.async_resolve_media",
return_value=PlayMedia(
url="http://example.com/test.png",
mime_type="image/png",
),
),
):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
@@ -136,53 +145,14 @@ async def test_image_url(
assert body == expected_data
@pytest.mark.usefixtures("mock_media_source")
async def test_no_images(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test when there are no images in the media folder."""
with patch(
"homeassistant.components.collection_image.image.async_browse_media",
return_value=BrowseMediaSource(
domain=None,
identifier=None,
media_class="",
media_content_type="",
title="",
can_play=False,
can_expand=True,
children=[],
),
):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(DEFAULT_ENTITY_ID)
assert state and state.state == STATE_UNAVAILABLE
await hass.async_block_till_done(wait_background_tasks=True)
assert (
"image.random_image: No valid images in media-source://mymedia" in caplog.text
)
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
async def test_media_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
config_entry: MockConfigEntry,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test when media browse throws an error."""
config_entry = config_entry_from_uri(MOCK_MEDIA_DIR_URI_EMPTY)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
@@ -193,7 +163,36 @@ async def test_media_error(
await hass.async_block_till_done(wait_background_tasks=True)
assert "image.random_image: Media Source not loaded" in caplog.text
assert (
f"image.random_image: No valid images in {MOCK_MEDIA_DIR_URI_EMPTY}"
in caplog.text
)
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR
@pytest.mark.usefixtures("mock_media_source")
async def test_media_error(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test when media browse throws an error."""
config_entry = config_entry_from_uri(MOCK_MEDIA_DIR_URI_BROWSE_ERROR)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
state = hass.states.get(DEFAULT_ENTITY_ID)
assert state and state.state == STATE_UNAVAILABLE
await hass.async_block_till_done(wait_background_tasks=True)
assert "image.random_image: Mock directory failed to browse" in caplog.text
client = await hass_client()
resp = await client.get(f"/api/image_proxy/{DEFAULT_ENTITY_ID}")
@@ -203,20 +202,22 @@ async def test_media_error(
async def test_unresolvable(
hass: HomeAssistant,
config_entry: MockConfigEntry,
browse_media_result: BrowseMediaSource,
media_source_state: MediaSourceState,
mock_media_source: MediaSourceMocks,
caplog: pytest.LogCaptureFixture,
hass_client: ClientSessionGenerator,
) -> None:
"""Test when resolving an image fails."""
media_source_state.resolve_exceptions[MOCK_MEDIA_IMAGE_URI_1] = Unresolvable(
"Mock image failed to resolve"
)
with (
patch(
"homeassistant.components.collection_image.image.async_browse_media",
return_value=browse_media_result,
),
):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert mock_media_source.image_browse.call_count == 1
assert mock_media_source.resolve.call_count == 1
state = hass.states.get(DEFAULT_ENTITY_ID)
@@ -224,32 +225,48 @@ async def test_unresolvable(
await hass.async_block_till_done(wait_background_tasks=True)
assert "image.random_image: Media Source not loaded" in caplog.text
assert "image.random_image: Mock image failed to resolve" in caplog.text
# Test we can recover by calling shuffle again when the image is resolvable
del media_source_state.resolve_exceptions[MOCK_MEDIA_IMAGE_URI_1]
with (
freeze_time(TEST_TIME),
):
await hass.services.async_call(
DOMAIN,
"shuffle",
{ATTR_ENTITY_ID: DEFAULT_ENTITY_ID},
blocking=True,
)
assert mock_media_source.image_browse.call_count == 2
assert mock_media_source.resolve.call_count == 2
state = hass.states.get(DEFAULT_ENTITY_ID)
assert state and state.state == TEST_TIME
await _verify_path_image(hass, hass_client)
@pytest.mark.usefixtures("mock_media_source")
async def test_image_file_read_error(
hass: HomeAssistant,
config_entry: MockConfigEntry,
browse_media_result: BrowseMediaSource,
media_source_state: MediaSourceState,
hass_client: ClientSessionGenerator,
) -> None:
"""Test that a file read error is surfaced when serving the image."""
missing_path = Path(__file__).parent / "does_not_exist.png"
media_source_state.resolve_results[MOCK_MEDIA_IMAGE_URI_1] = PlayMedia(
url="",
mime_type="image/png",
path=missing_path,
)
with (
freeze_time(TEST_TIME),
patch(
"homeassistant.components.collection_image.image.async_browse_media",
return_value=browse_media_result,
),
patch(
"homeassistant.components.collection_image.image.async_resolve_media",
return_value=PlayMedia(
url="",
mime_type="image/png",
path=missing_path,
),
),
):
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)