Impove LG webOS TV tests quality (#135130)

* Impove LG webOS TV tests quality

* Review comments
pull/135148/head
Shay Levy 2025-01-08 23:12:09 +02:00 committed by GitHub
parent 488c5a6b9f
commit bb4a497247
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 208 additions and 164 deletions

View File

@ -8,9 +8,7 @@ rules:
common-modules:
status: exempt
comment: The integration does not use common patterns.
config-flow-test-coverage:
status: todo
comment: remove duplicated config flow start in tests, make sure tests ends with CREATE_ENTRY or ABORT, use hass.config_entries.async_setup instead of async_setup_component, snapshot in diagnostics (and other tests when possible), test_client_disconnected validate no error in log
config-flow-test-coverage: done
config-flow:
status: todo
comment: make reauth flow more graceful
@ -39,7 +37,7 @@ rules:
log-when-unavailable: todo
parallel-updates: done
reauthentication-flow: done
test-coverage: todo
test-coverage: done
# Gold
devices: done

View File

@ -3,7 +3,6 @@
from homeassistant.components.webostv.const import DOMAIN
from homeassistant.const import CONF_CLIENT_SECRET, CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .const import CLIENT_KEY, FAKE_UUID, HOST, TV_NAME
@ -25,11 +24,7 @@ async def setup_webostv(
)
entry.add_to_hass(hass)
await async_setup_component(
hass,
DOMAIN,
{DOMAIN: {CONF_HOST: HOST}},
)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

View File

@ -0,0 +1,66 @@
# serializer version: 1
# name: test_diagnostics
dict({
'client': dict({
'apps': dict({
'com.webos.app.livetv': dict({
'icon': '**REDACTED**',
'id': 'com.webos.app.livetv',
'largeIcon': '**REDACTED**',
'title': 'Live TV',
}),
}),
'current_app_id': 'com.webos.app.livetv',
'current_channel': dict({
'channelId': 'ch1id',
'channelName': 'Channel 1',
'channelNumber': '1',
}),
'hello_info': dict({
'deviceUUID': '**REDACTED**',
}),
'inputs': dict({
'in1': dict({
'appId': 'app0',
'id': 'in1',
'label': 'Input01',
}),
'in2': dict({
'appId': 'app1',
'id': 'in2',
'label': 'Input02',
}),
}),
'is_connected': True,
'is_on': True,
'is_registered': True,
'software_info': dict({
'major_ver': 'major',
'minor_ver': 'minor',
}),
'sound_output': 'speaker',
'system_info': dict({
'modelName': 'MODEL',
}),
}),
'entry': dict({
'data': dict({
'client_secret': '**REDACTED**',
'host': '**REDACTED**',
}),
'disabled_by': None,
'discovery_keys': dict({
}),
'domain': 'webostv',
'minor_version': 1,
'options': dict({
}),
'pref_disable_new_entities': False,
'pref_disable_polling': False,
'source': 'user',
'title': 'LG webOS TV MODEL',
'unique_id': '**REDACTED**',
'version': 1,
}),
})
# ---

View File

@ -0,0 +1,59 @@
# serializer version: 1
# name: test_entity_attributes
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'tv',
'friendly_name': 'LG webOS TV MODEL',
'is_volume_muted': False,
'media_content_type': <MediaType.CHANNEL: 'channel'>,
'media_title': 'Channel 1',
'sound_output': 'speaker',
'source': 'Live TV',
'source_list': list([
'Input01',
'Input02',
'Live TV',
]),
'supported_features': <MediaPlayerEntityFeature: 24381>,
'volume_level': 0.37,
}),
'context': <ANY>,
'entity_id': 'media_player.lg_webos_tv_model',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_entity_attributes.1
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'configuration_url': None,
'connections': set({
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'webostv',
'some-fake-uuid',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'LG',
'model': 'MODEL',
'model_id': None,
'name': 'LG webOS TV MODEL',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'suggested_area': None,
'sw_version': 'major.minor',
'via_device_id': None,
})
# ---

View File

@ -1,7 +1,6 @@
"""Test the WebOS Tv config flow."""
import dataclasses
from unittest.mock import Mock
from unittest.mock import AsyncMock
from aiowebostv import WebOsTvPairError
import pytest
@ -41,28 +40,7 @@ MOCK_DISCOVERY_INFO = ssdp.SsdpServiceInfo(
async def test_form(hass: HomeAssistant, client) -> None:
"""Test we get the form."""
assert client
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_USER},
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_USER},
data=MOCK_USER_CONFIG,
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pairing"
"""Test successful user flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_USER},
@ -77,10 +55,10 @@ async def test_form(hass: HomeAssistant, client) -> None:
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TV_NAME
config_entry = result["result"]
assert config_entry.unique_id == FAKE_UUID
@pytest.mark.parametrize(
@ -114,27 +92,44 @@ async def test_options_flow_live_tv_in_apps(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result2 = await hass.config_entries.options.async_configure(
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_SOURCES: ["Live TV", "Input01", "Input02"]},
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["data"][CONF_SOURCES] == ["Live TV", "Input01", "Input02"]
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["data"][CONF_SOURCES] == ["Live TV", "Input01", "Input02"]
async def test_options_flow_cannot_retrieve(hass: HomeAssistant, client) -> None:
"""Test options config flow cannot retrieve sources."""
entry = await setup_webostv(hass)
client.connect = Mock(side_effect=ConnectionRefusedError())
client.connect = AsyncMock(side_effect=ConnectionRefusedError())
result = await hass.config_entries.options.async_init(entry.entry_id)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_retrieve"}
# recover
client.connect = AsyncMock(return_value=True)
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input=None,
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result3 = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={CONF_SOURCES: ["Input01", "Input02"]},
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
assert result3["data"][CONF_SOURCES] == ["Input01", "Input02"]
async def test_form_cannot_connect(hass: HomeAssistant, client) -> None:
"""Test we handle cannot connect error."""
@ -144,14 +139,22 @@ async def test_form_cannot_connect(hass: HomeAssistant, client) -> None:
data=MOCK_USER_CONFIG,
)
client.connect = Mock(side_effect=ConnectionRefusedError())
result2 = await hass.config_entries.flow.async_configure(
client.connect = AsyncMock(side_effect=ConnectionRefusedError())
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {"base": "cannot_connect"}
# recover
client.connect = AsyncMock(return_value=True)
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TV_NAME
async def test_form_pairexception(hass: HomeAssistant, client) -> None:
@ -162,20 +165,18 @@ async def test_form_pairexception(hass: HomeAssistant, client) -> None:
data=MOCK_USER_CONFIG,
)
client.connect = Mock(side_effect=WebOsTvPairError("error"))
result2 = await hass.config_entries.flow.async_configure(
client.connect = AsyncMock(side_effect=WebOsTvPairError("error"))
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "error_pairing"
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "error_pairing"
async def test_entry_already_configured(hass: HomeAssistant, client) -> None:
"""Test entry already configured."""
await setup_webostv(hass)
assert client
result = await hass.config_entries.flow.async_init(
DOMAIN,
@ -189,8 +190,6 @@ async def test_entry_already_configured(hass: HomeAssistant, client) -> None:
async def test_form_ssdp(hass: HomeAssistant, client) -> None:
"""Test that the ssdp confirmation form is served."""
assert client
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVERY_INFO
)
@ -199,19 +198,18 @@ async def test_form_ssdp(hass: HomeAssistant, client) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pairing"
result2 = await hass.config_entries.flow.async_configure(
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == TV_NAME
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TV_NAME
config_entry = result["result"]
assert config_entry.unique_id == FAKE_UUID
async def test_ssdp_in_progress(hass: HomeAssistant, client) -> None:
"""Test abort if ssdp paring is already in progress."""
assert client
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_USER},
@ -222,38 +220,19 @@ async def test_ssdp_in_progress(hass: HomeAssistant, client) -> None:
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "pairing"
result2 = await hass.config_entries.flow.async_init(
# Start another ssdp flow to make sure it aborts as already in progress
result = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=MOCK_DISCOVERY_INFO
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_in_progress"
async def test_ssdp_not_update_uuid(hass: HomeAssistant, client) -> None:
"""Test that ssdp not updates different host."""
entry = await setup_webostv(hass, None)
assert client
assert entry.unique_id is None
discovery_info = dataclasses.replace(MOCK_DISCOVERY_INFO)
discovery_info.ssdp_location = "http://1.2.3.5"
result2 = await hass.config_entries.flow.async_init(
DOMAIN, context={CONF_SOURCE: SOURCE_SSDP}, data=discovery_info
)
await hass.async_block_till_done()
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "pairing"
assert entry.unique_id is None
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_in_progress"
async def test_form_abort_uuid_configured(hass: HomeAssistant, client) -> None:
"""Test abort if uuid is already configured, verify host update."""
entry = await setup_webostv(hass, MOCK_DISCOVERY_INFO.upnp[ssdp.ATTR_UPNP_UDN][5:])
assert client
assert entry.unique_id == MOCK_DISCOVERY_INFO.upnp[ssdp.ATTR_UPNP_UDN][5:]
assert entry.data[CONF_HOST] == HOST
@ -268,6 +247,7 @@ async def test_form_abort_uuid_configured(hass: HomeAssistant, client) -> None:
user_config = {CONF_HOST: "new_host"}
# Start another flow to make sure it aborts and updates host
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_USER},
@ -282,8 +262,6 @@ async def test_form_abort_uuid_configured(hass: HomeAssistant, client) -> None:
result["flow_id"], user_input={}
)
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
assert entry.data[CONF_HOST] == "new_host"
@ -294,7 +272,6 @@ async def test_reauth_successful(
) -> None:
"""Test that the reauthorization is successful."""
entry = await setup_webostv(hass)
assert client
result = await entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
@ -327,7 +304,6 @@ async def test_reauth_errors(
) -> None:
"""Test reauthorization errors."""
entry = await setup_webostv(hass)
assert client
result = await entry.start_reauth_flow(hass)
assert result["step_id"] == "reauth_confirm"
@ -337,7 +313,7 @@ async def test_reauth_errors(
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
monkeypatch.setattr(client, "connect", Mock(side_effect=side_effect))
client.connect.side_effect = side_effect()
result = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input={}
)

View File

@ -1,6 +1,8 @@
"""Tests for the diagnostics data provided by LG webOS Smart TV."""
from homeassistant.components.diagnostics import REDACTED
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
from homeassistant.core import HomeAssistant
from . import setup_webostv
@ -10,56 +12,13 @@ from tests.typing import ClientSessionGenerator
async def test_diagnostics(
hass: HomeAssistant, hass_client: ClientSessionGenerator, client
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
client,
snapshot: SnapshotAssertion,
) -> None:
"""Test diagnostics."""
entry = await setup_webostv(hass)
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == {
"client": {
"is_registered": True,
"is_connected": True,
"current_app_id": "com.webos.app.livetv",
"current_channel": {
"channelId": "ch1id",
"channelName": "Channel 1",
"channelNumber": "1",
},
"apps": {
"com.webos.app.livetv": {
"icon": REDACTED,
"id": "com.webos.app.livetv",
"largeIcon": REDACTED,
"title": "Live TV",
}
},
"inputs": {
"in1": {"appId": "app0", "id": "in1", "label": "Input01"},
"in2": {"appId": "app1", "id": "in2", "label": "Input02"},
},
"system_info": {"modelName": "MODEL"},
"software_info": {"major_ver": "major", "minor_ver": "minor"},
"hello_info": {"deviceUUID": "**REDACTED**"},
"sound_output": "speaker",
"is_on": True,
},
"entry": {
"entry_id": entry.entry_id,
"version": 1,
"minor_version": 1,
"domain": "webostv",
"title": "LG webOS TV MODEL",
"data": {
"client_secret": "**REDACTED**",
"host": "**REDACTED**",
},
"options": {},
"pref_disable_new_entities": False,
"pref_disable_polling": False,
"source": "user",
"unique_id": REDACTED,
"disabled_by": None,
"created_at": entry.created_at.isoformat(),
"modified_at": entry.modified_at.isoformat(),
"discovery_keys": {},
},
}
assert await get_diagnostics_for_config_entry(hass, hass_client, entry) == snapshot(
exclude=props("created_at", "modified_at", "entry_id")
)

View File

@ -5,7 +5,10 @@ from http import HTTPStatus
from unittest.mock import Mock
from aiowebostv import WebOsTvPairError
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from syrupy.filters import props
from homeassistant.components import automation
from homeassistant.components.media_player import (
@ -19,7 +22,6 @@ from homeassistant.components.media_player import (
DOMAIN as MP_DOMAIN,
SERVICE_PLAY_MEDIA,
SERVICE_SELECT_SOURCE,
MediaPlayerDeviceClass,
MediaPlayerEntityFeature,
MediaPlayerState,
MediaType,
@ -42,7 +44,6 @@ from homeassistant.components.webostv.media_player import (
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import (
ATTR_COMMAND,
ATTR_DEVICE_CLASS,
ATTR_ENTITY_ID,
ATTR_SUPPORTED_FEATURES,
ENTITY_MATCH_NONE,
@ -58,7 +59,6 @@ from homeassistant.const import (
SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP,
STATE_OFF,
STATE_ON,
)
from homeassistant.core import HomeAssistant, State
from homeassistant.exceptions import HomeAssistantError
@ -67,7 +67,7 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import setup_webostv
from .const import CHANNEL_2, ENTITY_ID, TV_MODEL, TV_NAME
from .const import CHANNEL_2, ENTITY_ID, TV_NAME
from tests.common import async_fire_time_changed, mock_restore_cache
from tests.test_util.aiohttp import AiohttpClientMocker
@ -298,6 +298,7 @@ async def test_entity_attributes(
client,
monkeypatch: pytest.MonkeyPatch,
device_registry: dr.DeviceRegistry,
snapshot: SnapshotAssertion,
) -> None:
"""Test entity attributes."""
entry = await setup_webostv(hass)
@ -305,18 +306,7 @@ async def test_entity_attributes(
# Attributes when device is on
state = hass.states.get(ENTITY_ID)
attrs = state.attributes
assert state.state == STATE_ON
assert state.name == TV_NAME
assert attrs[ATTR_DEVICE_CLASS] == MediaPlayerDeviceClass.TV
assert attrs[ATTR_MEDIA_VOLUME_MUTED] is False
assert attrs[ATTR_MEDIA_VOLUME_LEVEL] == 0.37
assert attrs[ATTR_INPUT_SOURCE] == "Live TV"
assert attrs[ATTR_INPUT_SOURCE_LIST] == ["Input01", "Input02", "Live TV"]
assert attrs[ATTR_MEDIA_CONTENT_TYPE] == MediaType.CHANNEL
assert attrs[ATTR_MEDIA_TITLE] == "Channel 1"
assert attrs[ATTR_SOUND_OUTPUT] == "speaker"
assert state == snapshot(exclude=props("entity_picture"))
# Volume level not available
monkeypatch.setattr(client, "volume", None)
@ -334,13 +324,7 @@ async def test_entity_attributes(
# Device Info
device = device_registry.async_get_device(identifiers={(DOMAIN, entry.unique_id)})
assert device
assert device.identifiers == {(DOMAIN, entry.unique_id)}
assert device.manufacturer == "LG"
assert device.name == TV_NAME
assert device.sw_version == "major.minor"
assert device.model == TV_MODEL
assert device == snapshot
# Sound output when off
monkeypatch.setattr(client, "sound_output", None)
@ -473,16 +457,23 @@ async def test_update_sources_live_tv_find(
async def test_client_disconnected(
hass: HomeAssistant, client, monkeypatch: pytest.MonkeyPatch
hass: HomeAssistant,
client,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test error not raised when client is disconnected."""
await setup_webostv(hass)
monkeypatch.setattr(client, "is_connected", Mock(return_value=False))
monkeypatch.setattr(client, "connect", Mock(side_effect=TimeoutError))
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=20))
freezer.tick(timedelta(seconds=20))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert "TimeoutError" not in caplog.text
async def test_control_error_handling(
hass: HomeAssistant,