Avoid swallowing exceptions in arcam_fmj services (#170681)

Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
This commit is contained in:
Joakim Plate
2026-05-14 20:41:18 -04:00
committed by GitHub
co-authored by Paulus Schoutsen
parent 6d10249a93
commit 20d7aa5523
3 changed files with 129 additions and 20 deletions
@@ -18,10 +18,10 @@ from homeassistant.components.media_player import (
)
from homeassistant.const import ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import EVENT_TURN_ON
from .const import DOMAIN, EVENT_TURN_ON
from .coordinator import ArcamFmjConfigEntry, ArcamFmjCoordinator
from .entity import ArcamFmjEntity
@@ -52,7 +52,7 @@ def convert_exception[**_P, _R](
return await func(*args, **kwargs)
except ConnectionFailed as exception:
raise HomeAssistantError(
f"Connection failed to device during {func}"
translation_domain=DOMAIN, translation_key="connection_failed"
) from exception
return _convert_exception
@@ -96,10 +96,12 @@ class ArcamFmj(ArcamFmjEntity, MediaPlayerEntity):
"""Select a specific source."""
try:
value = SourceCodes[source]
# pylint: disable-next=home-assistant-action-swallowed-exception
except KeyError:
_LOGGER.error("Unsupported source %s", source)
return
except KeyError as exception:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unsupported_source",
translation_placeholders={"source": source},
) from exception
await self._state.set_source(value)
self.async_write_ha_state()
@@ -110,8 +112,10 @@ class ArcamFmj(ArcamFmjEntity, MediaPlayerEntity):
try:
await self._state.set_decode_mode(sound_mode)
except (KeyError, ValueError) as exception:
raise HomeAssistantError(
f"Unsupported sound_mode {sound_mode}"
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unsupported_sound_mode",
translation_placeholders={"sound_mode": sound_mode},
) from exception
self.async_write_ha_state()
@@ -194,8 +198,11 @@ class ArcamFmj(ArcamFmjEntity, MediaPlayerEntity):
preset = int(media_id[7:])
await self._state.set_tuner_preset(preset)
else:
_LOGGER.error("Media %s is not supported", media_id)
return
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="unsupported_media",
translation_placeholders={"media": media_id},
)
@property
def source(self) -> str | None:
@@ -139,5 +139,19 @@
"name": "Incoming video vertical resolution"
}
}
},
"exceptions": {
"connection_failed": {
"message": "Connection failed to the device."
},
"unsupported_media": {
"message": "Unsupported media: {media}."
},
"unsupported_sound_mode": {
"message": "Unsupported sound mode: {sound_mode}."
},
"unsupported_source": {
"message": "Unsupported source: {source}."
}
}
}
@@ -17,12 +17,14 @@ from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_ARTIST,
ATTR_MEDIA_CHANNEL,
ATTR_MEDIA_CONTENT_ID,
ATTR_MEDIA_CONTENT_TYPE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
ATTR_SOUND_MODE,
ATTR_SOUND_MODE_LIST,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOUND_MODE,
SERVICE_SELECT_SOURCE,
SERVICE_TURN_OFF,
@@ -35,7 +37,7 @@ from homeassistant.components.media_player import (
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant, State as CoreState
from homeassistant.exceptions import HomeAssistantError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from .conftest import MOCK_ENTITY_ID
@@ -187,14 +189,14 @@ async def test_update_lost(
@pytest.mark.parametrize(
("source", "value"),
[("PVR", SourceCodes.PVR), ("BD", SourceCodes.BD), ("INVALID", None)],
[("PVR", SourceCodes.PVR), ("BD", SourceCodes.BD)],
)
@pytest.mark.usefixtures("player_setup")
async def test_select_source(
async def test_select_valid_source(
hass: HomeAssistant,
state_1: State,
source: str,
value: SourceCodes | None,
value: SourceCodes,
) -> None:
"""Test selection of source."""
await hass.services.async_call(
@@ -204,10 +206,29 @@ async def test_select_source(
blocking=True,
)
if value:
state_1.set_source.assert_called_with(value)
else:
state_1.set_source.assert_not_called()
state_1.set_source.assert_called_with(value)
@pytest.mark.usefixtures("player_setup")
async def test_select_invalid_source(
hass: HomeAssistant,
state_1: State,
) -> None:
"""Test selection of source."""
with pytest.raises(
ServiceValidationError,
check=lambda e: (
e.translation_domain == "arcam_fmj"
and e.translation_key == "unsupported_source"
),
):
await hass.services.async_call(
"media_player",
SERVICE_SELECT_SOURCE,
service_data={ATTR_ENTITY_ID: MOCK_ENTITY_ID, ATTR_INPUT_SOURCE: "INVALID"},
blocking=True,
)
state_1.set_source.assert_not_called()
@pytest.mark.usefixtures("player_setup")
@@ -239,6 +260,28 @@ async def test_select_sound_mode(
state_1.set_decode_mode.assert_called_with(mode)
@pytest.mark.usefixtures("player_setup")
async def test_select_invalid_sound_mode(
hass: HomeAssistant,
state_1: State,
) -> None:
"""Test selection of source."""
state_1.set_decode_mode.side_effect = KeyError()
with pytest.raises(
ServiceValidationError,
check=lambda e: (
e.translation_domain == "arcam_fmj"
and e.translation_key == "unsupported_sound_mode"
),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOUND_MODE,
service_data={ATTR_ENTITY_ID: MOCK_ENTITY_ID, ATTR_SOUND_MODE: "INVALID"},
blocking=True,
)
@pytest.mark.usefixtures("player_setup")
async def test_volume_up(hass: HomeAssistant, state_1: State) -> None:
"""Test mute functionality."""
@@ -263,6 +306,45 @@ async def test_volume_down(hass: HomeAssistant, state_1: State) -> None:
state_1.dec_volume.assert_called_with()
@pytest.mark.usefixtures("player_setup")
async def test_play_media(hass: HomeAssistant, state_1: State) -> None:
"""Test mute functionality."""
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
service_data={
ATTR_ENTITY_ID: MOCK_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "preset:1",
},
blocking=True,
)
state_1.set_tuner_preset.assert_called_with(1)
@pytest.mark.usefixtures("player_setup")
async def test_play_media_invalid(hass: HomeAssistant, state_1: State) -> None:
"""Test mute functionality."""
with pytest.raises(
ServiceValidationError,
check=lambda e: (
e.translation_domain == "arcam_fmj"
and e.translation_key == "unsupported_media"
),
):
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_PLAY_MEDIA,
service_data={
ATTR_ENTITY_ID: MOCK_ENTITY_ID,
ATTR_MEDIA_CONTENT_TYPE: MediaType.MUSIC,
ATTR_MEDIA_CONTENT_ID: "invalid",
},
blocking=True,
)
state_1.set_tuner_preset.assert_not_called()
@pytest.mark.parametrize(
("mode", "mode_enum"),
[
@@ -360,7 +442,13 @@ async def test_set_volume_level_lost(hass: HomeAssistant, state_1: State) -> Non
state_1.set_volume.side_effect = ConnectionFailed()
with pytest.raises(HomeAssistantError):
with pytest.raises(
HomeAssistantError,
check=lambda e: (
e.translation_domain == "arcam_fmj"
and e.translation_key == "connection_failed"
),
):
await hass.services.async_call(
"media_player",
SERVICE_VOLUME_SET,