Rework enigma2 tests (#135475)

This commit is contained in:
Sid 2025-01-19 13:24:47 +01:00 committed by GitHub
parent 3978c4cdb3
commit a55bd593af
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 808 additions and 116 deletions

View File

@ -56,6 +56,7 @@ class Enigma2Device(CoordinatorEntity[Enigma2UpdateCoordinator], MediaPlayerEnti
| MediaPlayerEntityFeature.TURN_ON
| MediaPlayerEntityFeature.PAUSE
| MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.PLAY
)
def __init__(self, coordinator: Enigma2UpdateCoordinator) -> None:

View File

@ -1,6 +1,10 @@
"""Test the Enigma2 config flow."""
from openwebif.api import OpenWebIfServiceEvent, OpenWebIfStatus
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
from openwebif.api import OpenWebIfDevice, OpenWebIfServiceEvent, OpenWebIfStatus
import pytest
from homeassistant.components.enigma2.const import (
CONF_DEEP_STANDBY,
@ -10,6 +14,7 @@ from homeassistant.components.enigma2.const import (
DEFAULT_PORT,
DEFAULT_SSL,
DEFAULT_VERIFY_SSL,
DOMAIN,
)
from homeassistant.const import (
CONF_HOST,
@ -20,6 +25,8 @@ from homeassistant.const import (
CONF_VERIFY_SSL,
)
from tests.common import MockConfigEntry, load_json_object_fixture
MAC_ADDRESS = "12:34:56:78:90:ab"
TEST_REQUIRED = {
@ -45,42 +52,41 @@ EXPECTED_OPTIONS = {
}
class MockDevice:
"""A mock Enigma2 device."""
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN, data=TEST_REQUIRED, unique_id="12:34:56:78:90:ab"
)
mac_address: str | None = "12:34:56:78:90:ab"
_base = "http://1.1.1.1"
def __init__(self) -> None:
"""Initialize the mock Enigma2 device."""
self.status = OpenWebIfStatus(currservice=OpenWebIfServiceEvent())
@pytest.fixture
def openwebif_device_mock() -> Generator[AsyncMock]:
"""Mock a OpenWebIf device."""
async def _call_api(self, url: str) -> dict | None:
if url.endswith("/api/about"):
return {
"info": {
"ifaces": [
{
"mac": self.mac_address,
}
],
"model": "Mock Enigma2",
"brand": "Enigma2",
}
}
return None
def get_version(self) -> str | None:
"""Return the version."""
return None
async def get_about(self) -> dict:
"""Get mock about endpoint."""
return await self._call_api("/api/about")
async def get_all_bouquets(self) -> dict:
"""Get all bouquets."""
return {
with (
patch(
"homeassistant.components.enigma2.coordinator.OpenWebIfDevice",
spec=OpenWebIfDevice,
) as openwebif_device_mock,
patch(
"homeassistant.components.enigma2.config_flow.OpenWebIfDevice",
new=openwebif_device_mock,
),
):
device = openwebif_device_mock.return_value
device.status = OpenWebIfStatus(currservice=OpenWebIfServiceEvent())
device.turn_off_to_deep = False
device.sources = {"Test": "1"}
device.source_list = list(device.sources.keys())
device.picon_url = "file:///"
device.get_about.return_value = load_json_object_fixture(
"device_about.json", DOMAIN
)
device.get_status_info.return_value = load_json_object_fixture(
"device_statusinfo_on.json", DOMAIN
)
device.get_all_bouquets.return_value = {
"bouquets": [
[
'1:7:1:0:0:0:0:0:0:0:FROM BOUQUET "userbouquet.favourites.tv" ORDER BY bouquet',
@ -88,9 +94,4 @@ class MockDevice:
]
]
}
async def update(self) -> None:
"""Mock update."""
async def close(self):
"""Mock close."""
yield device

View File

@ -0,0 +1,158 @@
{
"info": {
"brand": "GigaBlue",
"model": "UHD QUAD 4K",
"boxtype": "gbquad4k",
"machinebuild": "gb7252",
"lcd": 1,
"grabpip": 1,
"chipset": "bcm7252s",
"mem1": "906132 kB",
"mem2": "616396 kB",
"mem3": "616396 kB frei / 906132 kB insgesamt",
"uptime": "46d 15:47",
"webifver": "OWIF 2.2.0",
"imagedistro": "openatv",
"friendlyimagedistro": "openATV",
"oever": "OE-Alliance 5.5",
"imagever": "7.5.20241101",
"enigmaver": "2024-10-31",
"driverdate": "20200723",
"kernelver": "4.1.20",
"fp_version": 0,
"friendlychipsetdescription": "Chipsatz",
"friendlychipsettext": "Broadcom 7252s",
"tuners": [
{
"name": "Tuner A",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner B",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner C",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner D",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner E",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner F",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner G",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner H",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner I",
"type": "GIGA DVB-T2/C NIM (TT2L10) (DVB-T2)",
"rec": "",
"live": "",
"stream": ""
}
],
"ifaces": [
{
"name": "eth0",
"friendlynic": "Broadcom Gigabit Ethernet",
"linkspeed": "1 GBit/s",
"mac": "12:34:56:78:90:ab",
"dhcp": true,
"ipv4method": "DHCP",
"ip": "192.168.1.100",
"mask": "255.255.255.0",
"v4prefix": 23,
"gw": "192.168.1.1",
"ipv6": "2003::2/64",
"ipmethod": "SL-AAC",
"firstpublic": "2003::2"
}
],
"hdd": [
{
"model": "ATA(ST2000LM015-2E81)",
"capacity": "1.8 TB",
"labelled_capacity": "2.0 TB",
"free": "22.5 GB",
"mount": "/media/hdd",
"friendlycapacity": "22.5 GB frei / 1.8 TB (2.0 TB) insgesamt"
}
],
"shares": [
{
"name": "NAS",
"method": "autofs",
"type": "SMBv2.0",
"mode": "r/w",
"path": "//192.168.1.2/NAS",
"host": "192.168.1.2",
"ipaddress": null,
"friendlyaddress": "192.168.1.2"
}
],
"transcoding": true,
"EX": "",
"streams": [],
"timerpipzap": false,
"allow_duplicate": false,
"timermargins": true,
"textinputsupport": true
},
"service": {
"result": false,
"name": "",
"namespace": "",
"aspect": 0,
"provider": "",
"width": 0,
"height": 0,
"apid": 0,
"vpid": 0,
"pcrpid": 0,
"pmtpid": 0,
"txtpid": "N/A",
"tsid": 0,
"onid": 0,
"sid": 0,
"ref": "",
"iswidescreen": false,
"bqref": "",
"bqname": ""
}
}

View File

@ -0,0 +1,158 @@
{
"info": {
"brand": "GigaBlue",
"model": "UHD QUAD 4K",
"boxtype": "gbquad4k",
"machinebuild": "gb7252",
"lcd": 1,
"grabpip": 1,
"chipset": "bcm7252s",
"mem1": "906132 kB",
"mem2": "616396 kB",
"mem3": "616396 kB frei / 906132 kB insgesamt",
"uptime": "46d 15:47",
"webifver": "OWIF 2.2.0",
"imagedistro": "openatv",
"friendlyimagedistro": "openATV",
"oever": "OE-Alliance 5.5",
"imagever": "7.5.20241101",
"enigmaver": "2024-10-31",
"driverdate": "20200723",
"kernelver": "4.1.20",
"fp_version": 0,
"friendlychipsetdescription": "Chipsatz",
"friendlychipsettext": "Broadcom 7252s",
"tuners": [
{
"name": "Tuner A",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner B",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner C",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner D",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner E",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner F",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner G",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner H",
"type": "DVB-S2X NIM(45308X FBC) (DVB-S2X)",
"rec": "",
"live": "",
"stream": ""
},
{
"name": "Tuner I",
"type": "GIGA DVB-T2/C NIM (TT2L10) (DVB-T2)",
"rec": "",
"live": "",
"stream": ""
}
],
"ifaces": [
{
"name": "eth0",
"friendlynic": "Broadcom Gigabit Ethernet",
"linkspeed": "1 GBit/s",
"mac": null,
"dhcp": true,
"ipv4method": "DHCP",
"ip": "192.168.1.100",
"mask": "255.255.255.0",
"v4prefix": 23,
"gw": "192.168.1.1",
"ipv6": "2003::2/64",
"ipmethod": "SL-AAC",
"firstpublic": "2003::2"
}
],
"hdd": [
{
"model": "ATA(ST2000LM015-2E81)",
"capacity": "1.8 TB",
"labelled_capacity": "2.0 TB",
"free": "22.5 GB",
"mount": "/media/hdd",
"friendlycapacity": "22.5 GB frei / 1.8 TB (2.0 TB) insgesamt"
}
],
"shares": [
{
"name": "NAS",
"method": "autofs",
"type": "SMBv2.0",
"mode": "r/w",
"path": "//192.168.1.2/NAS",
"host": "192.168.1.2",
"ipaddress": null,
"friendlyaddress": "192.168.1.2"
}
],
"transcoding": true,
"EX": "",
"streams": [],
"timerpipzap": false,
"allow_duplicate": false,
"timermargins": true,
"textinputsupport": true
},
"service": {
"result": false,
"name": "",
"namespace": "",
"aspect": 0,
"provider": "",
"width": 0,
"height": 0,
"apid": 0,
"vpid": 0,
"pcrpid": 0,
"pmtpid": 0,
"txtpid": "N/A",
"tsid": 0,
"onid": 0,
"sid": 0,
"ref": "",
"iswidescreen": false,
"bqref": "",
"bqname": ""
}
}

View File

@ -0,0 +1,20 @@
{
"volume": 100,
"muted": false,
"transcoding": true,
"currservice_filename": "",
"currservice_id": 38835,
"currservice_name": "Flucht aus Saudi-Arabien",
"currservice_serviceref": "1:0:19:2BA2:3F2:1:C00000:0:0:0:",
"currservice_begin": "16:30",
"currservice_begin_timestamp": 1734622200,
"currservice_end": "17:15",
"currservice_end_timestamp": 1734624900,
"currservice_description": "Ein M\u00e4dchen k\u00e4mpft um die Freiheit",
"currservice_station": "ZDFinfo HD",
"currservice_fulldescription": "Flucht aus Saudi-Arabien\n16:30 - 17:15\n\nSaudi-Arabien / Australien 2019\nIm streng islamischen K\u00f6nigreich Saudi-Arabien haben Frauen immer noch wenig Rechte. Wenn sie ein selbstbestimmtes Leben f\u00fchren wollen, bleibt ihnen h\u00e4ufig nur die Flucht.\n\nDie 18-j\u00e4hrige Rahaf Mohammed al-Qunun ist eine solche junge Frau, die riskiert hat, dem m\u00e4nnlich gepr\u00e4gten Vormundschaftssystem zu entfliehen. Doch der saudische Staat und Rahafs Familie verfolgen die Abtr\u00fcnnige sogar bis ins Ausland.\nHD-Produktion",
"inStandby": "false",
"isRecording": "false",
"Streaming_list": "",
"isStreaming": "false"
}

View File

@ -0,0 +1,16 @@
{
"volume": 100,
"muted": true,
"transcoding": true,
"currservice_filename": "",
"currservice_id": -1,
"currservice_name": "N/A",
"currservice_begin": "",
"currservice_end": "",
"currservice_description": "",
"currservice_fulldescription": "N/A",
"inStandby": "true",
"isRecording": "false",
"Streaming_list": "",
"isStreaming": "false"
}

View File

@ -1,19 +1,19 @@
"""Test the Enigma2 config flow."""
from typing import Any
from unittest.mock import patch
from unittest.mock import AsyncMock
from aiohttp.client_exceptions import ClientError
from openwebif.error import InvalidAuthError
import pytest
from homeassistant import config_entries
from homeassistant.components.enigma2.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from .conftest import TEST_FULL, TEST_REQUIRED, MockDevice
from .conftest import TEST_FULL, TEST_REQUIRED
from tests.common import MockConfigEntry
@ -22,42 +22,35 @@ from tests.common import MockConfigEntry
async def user_flow(hass: HomeAssistant) -> str:
"""Return a user-initiated flow after filling in host info."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] is None
return result["flow_id"]
@pytest.mark.usefixtures("openwebif_device_mock")
@pytest.mark.parametrize(
("test_config"),
[(TEST_FULL), (TEST_REQUIRED)],
)
async def test_form_user(
hass: HomeAssistant, user_flow: str, test_config: dict[str, Any]
) -> None:
async def test_form_user(hass: HomeAssistant, test_config: dict[str, Any]) -> None:
"""Test a successful user initiated flow."""
with (
patch(
"openwebif.api.OpenWebIfDevice.__new__",
return_value=MockDevice(),
),
patch(
"homeassistant.components.enigma2.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result = await hass.config_entries.flow.async_configure(user_flow, test_config)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "user"
result = await hass.config_entries.flow.async_configure(
result["flow_id"], test_config
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == test_config[CONF_HOST]
assert result["data"] == test_config
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "error_type"),
("side_effect", "error_value"),
[
(InvalidAuthError, "invalid_auth"),
(ClientError, "cannot_connect"),
@ -65,46 +58,87 @@ async def test_form_user(
],
)
async def test_form_user_errors(
hass: HomeAssistant, user_flow, exception: Exception, error_type: str
hass: HomeAssistant,
openwebif_device_mock: AsyncMock,
side_effect: Exception,
error_value: str,
) -> None:
"""Test we handle errors."""
with patch(
"homeassistant.components.enigma2.config_flow.OpenWebIfDevice.__new__",
side_effect=exception,
):
result = await hass.config_entries.flow.async_configure(user_flow, TEST_FULL)
openwebif_device_mock.get_about.side_effect = side_effect
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": 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_configure(
result["flow_id"], TEST_FULL
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == config_entries.SOURCE_USER
assert result["errors"] == {"base": error_type}
assert result["step_id"] == SOURCE_USER
assert result["errors"] == {"base": error_value}
openwebif_device_mock.get_about.side_effect = None
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
TEST_FULL,
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == TEST_FULL[CONF_HOST]
assert result["data"] == TEST_FULL
assert result["result"].unique_id == openwebif_device_mock.mac_address
async def test_options_flow(hass: HomeAssistant, user_flow: str) -> None:
@pytest.mark.usefixtures("openwebif_device_mock")
async def test_duplicate_host(
hass: HomeAssistant, mock_config_entry: MockConfigEntry
) -> None:
"""Test that a duplicate host aborts the config flow."""
mock_config_entry.add_to_hass(hass)
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
)
assert result2["type"] is FlowResultType.FORM
assert result2["step_id"] == "user"
result2 = await hass.config_entries.flow.async_configure(
result2["flow_id"], TEST_FULL
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "already_configured"
@pytest.mark.usefixtures("openwebif_device_mock")
async def test_options_flow(hass: HomeAssistant) -> None:
"""Test the form options."""
with patch(
"openwebif.api.OpenWebIfDevice.__new__",
return_value=MockDevice(),
):
entry = MockConfigEntry(domain=DOMAIN, data=TEST_FULL, options={}, entry_id="1")
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entry = MockConfigEntry(domain=DOMAIN, data=TEST_FULL, options={}, entry_id="1")
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is config_entries.ConfigEntryState.LOADED
assert entry.state is ConfigEntryState.LOADED
result = await hass.config_entries.options.async_init(entry.entry_id)
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={"source_bouquet": "Favourites (TV)"}
)
result = await hass.config_entries.options.async_configure(
result["flow_id"], user_input={"source_bouquet": "Favourites (TV)"}
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert entry.options == {"source_bouquet": "Favourites (TV)"}
assert result["type"] is FlowResultType.CREATE_ENTRY
assert entry.options == {"source_bouquet": "Favourites (TV)"}
await hass.async_block_till_done()
await hass.async_block_till_done()
assert entry.state is config_entries.ConfigEntryState.LOADED
assert entry.state is ConfigEntryState.LOADED

View File

@ -1,46 +1,45 @@
"""Test the Enigma2 integration init."""
from unittest.mock import patch
from unittest.mock import AsyncMock
import pytest
from homeassistant.components.enigma2.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from .conftest import TEST_REQUIRED, MockDevice
from .conftest import TEST_REQUIRED
from tests.common import MockConfigEntry
from tests.common import MockConfigEntry, load_json_object_fixture
async def test_device_without_mac_address(
hass: HomeAssistant, device_registry: dr.DeviceRegistry
hass: HomeAssistant,
openwebif_device_mock: AsyncMock,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test that a device gets successfully registered when the device doesn't report a MAC address."""
mock_device = MockDevice()
mock_device.mac_address = None
with patch(
"homeassistant.components.enigma2.coordinator.OpenWebIfDevice.__new__",
return_value=mock_device,
):
entry = MockConfigEntry(
domain=DOMAIN, data=TEST_REQUIRED, title="name", unique_id="123456"
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert device_registry.async_get_device({(DOMAIN, entry.unique_id)}) is not None
openwebif_device_mock.get_about.return_value = load_json_object_fixture(
"device_about_without_mac.json", DOMAIN
)
entry = MockConfigEntry(
domain=DOMAIN, data=TEST_REQUIRED, title="name", unique_id="123456"
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.unique_id == "123456"
assert device_registry.async_get_device({(DOMAIN, entry.unique_id)}) is not None
@pytest.mark.usefixtures("openwebif_device_mock")
async def test_unload_entry(hass: HomeAssistant) -> None:
"""Test successful unload of entry."""
with patch(
"homeassistant.components.enigma2.coordinator.OpenWebIfDevice.__new__",
return_value=MockDevice(),
):
entry = MockConfigEntry(domain=DOMAIN, data=TEST_REQUIRED, title="name")
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
entry = MockConfigEntry(domain=DOMAIN, data=TEST_REQUIRED, title="name")
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state is ConfigEntryState.LOADED

View File

@ -0,0 +1,305 @@
"""Tests for the media player module."""
from datetime import timedelta
from unittest.mock import AsyncMock
from freezegun.api import FrozenDateTimeFactory
from openwebif.api import OpenWebIfServiceEvent, OpenWebIfStatus
from openwebif.enums import PowerState, RemoteControlCodes, SetVolumeOption
import pytest
from homeassistant.components.enigma2.const import DOMAIN
from homeassistant.components.enigma2.media_player import ATTR_MEDIA_CURRENTLY_RECORDING
from homeassistant.components.media_player import (
ATTR_INPUT_SOURCE,
ATTR_MEDIA_VOLUME_LEVEL,
ATTR_MEDIA_VOLUME_MUTED,
DOMAIN as MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
MediaPlayerState,
)
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_MEDIA_NEXT_TRACK,
SERVICE_MEDIA_PAUSE,
SERVICE_MEDIA_PLAY,
SERVICE_MEDIA_PREVIOUS_TRACK,
SERVICE_MEDIA_STOP,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
SERVICE_VOLUME_DOWN,
SERVICE_VOLUME_MUTE,
SERVICE_VOLUME_SET,
SERVICE_VOLUME_UP,
)
from homeassistant.core import HomeAssistant
from tests.common import (
MockConfigEntry,
async_fire_time_changed,
load_json_object_fixture,
)
@pytest.mark.parametrize(
("deep_standby", "powerstate"),
[(False, PowerState.STANDBY), (True, PowerState.DEEP_STANDBY)],
)
async def test_turn_off(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
deep_standby: bool,
powerstate: PowerState,
) -> None:
"""Test turning off the media player."""
openwebif_device_mock.turn_off_to_deep = deep_standby
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: "media_player.1_1_1_1"}
)
openwebif_device_mock.set_powerstate.assert_awaited_once_with(powerstate)
async def test_turn_on(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
) -> None:
"""Test turning on the media player."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: "media_player.1_1_1_1"}
)
openwebif_device_mock.turn_on.assert_awaited_once()
async def test_set_volume_level(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
) -> None:
"""Test setting the volume of the media player."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_SET,
{ATTR_ENTITY_ID: "media_player.1_1_1_1", ATTR_MEDIA_VOLUME_LEVEL: 0.2},
)
openwebif_device_mock.set_volume.assert_awaited_once_with(20)
async def test_volume_up(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
) -> None:
"""Test increasing the volume of the media player."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: "media_player.1_1_1_1"}
)
openwebif_device_mock.set_volume.assert_awaited_once_with(SetVolumeOption.UP)
async def test_volume_down(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
) -> None:
"""Test decreasing the volume of the media player."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_DOWN,
{ATTR_ENTITY_ID: "media_player.1_1_1_1"},
)
openwebif_device_mock.set_volume.assert_awaited_once_with(SetVolumeOption.DOWN)
@pytest.mark.parametrize(
("service", "remote_code"),
[
(SERVICE_MEDIA_STOP, RemoteControlCodes.STOP),
(SERVICE_MEDIA_PLAY, RemoteControlCodes.PLAY),
(SERVICE_MEDIA_PAUSE, RemoteControlCodes.PAUSE),
(SERVICE_MEDIA_NEXT_TRACK, RemoteControlCodes.CHANNEL_UP),
(SERVICE_MEDIA_PREVIOUS_TRACK, RemoteControlCodes.CHANNEL_DOWN),
],
)
async def test_remote_control_actions(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
service: str,
remote_code: RemoteControlCodes,
) -> None:
"""Test media stop."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
service,
{ATTR_ENTITY_ID: "media_player.1_1_1_1"},
)
openwebif_device_mock.send_remote_control_action.assert_awaited_once_with(
remote_code
)
@pytest.mark.parametrize("mute", [False, True])
async def test_volume_mute(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
mute: bool,
) -> None:
"""Test mute."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_VOLUME_MUTE,
{ATTR_ENTITY_ID: "media_player.1_1_1_1", ATTR_MEDIA_VOLUME_MUTED: mute},
)
openwebif_device_mock.toggle_mute.assert_awaited_once()
async def test_select_source(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
) -> None:
"""Test media previous track."""
openwebif_device_mock.return_value.sources = {"Test": "1"}
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
MEDIA_PLAYER_DOMAIN,
SERVICE_SELECT_SOURCE,
{ATTR_ENTITY_ID: "media_player.1_1_1_1", ATTR_INPUT_SOURCE: "Test"},
)
openwebif_device_mock.zap.assert_awaited_once_with("1")
async def test_update_data_standby(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test data handling."""
openwebif_device_mock.get_status_info.return_value = load_json_object_fixture(
"device_statusinfo_standby.json", DOMAIN
)
openwebif_device_mock.status = OpenWebIfStatus(
currservice=OpenWebIfServiceEvent(), in_standby=True
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
ATTR_MEDIA_CURRENTLY_RECORDING
not in hass.states.get("media_player.1_1_1_1").attributes
)
assert hass.states.get("media_player.1_1_1_1").state == MediaPlayerState.OFF
async def test_update_volume(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test volume data handling."""
openwebif_device_mock.status = OpenWebIfStatus(
currservice=OpenWebIfServiceEvent(), in_standby=False, volume=100
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
hass.states.get("media_player.1_1_1_1").attributes[ATTR_MEDIA_VOLUME_LEVEL]
> 0.99
)
async def test_update_volume_none(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
openwebif_device_mock: AsyncMock,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test volume data handling."""
openwebif_device_mock.status = OpenWebIfStatus(
currservice=OpenWebIfServiceEvent(), in_standby=False, volume=None
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
freezer.tick(timedelta(seconds=30))
async_fire_time_changed(hass)
await hass.async_block_till_done()
assert (
ATTR_MEDIA_VOLUME_LEVEL
not in hass.states.get("media_player.1_1_1_1").attributes
)