Add 100% coverage of Reolink button platform (#124380)

* Add 100% button coverage

* review comments

* fix

* Use SERVICE_PRESS constant

* Use DOMAIN instead of const.DOMAIN

* styling

* User entity_registry_enabled_by_default fixture

* fixes

* Split out ptz_move test

* use SERVICE_PTZ_MOVE constant
This commit is contained in:
starkillerOG 2024-09-04 12:16:57 +02:00 committed by GitHub
parent fb5afff9d5
commit 4b111008df
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 164 additions and 57 deletions

View File

@ -37,6 +37,7 @@ from .entity import (
ATTR_SPEED = "speed"
SUPPORT_PTZ_SPEED = CameraEntityFeature.STREAM
SERVICE_PTZ_MOVE = "ptz_move"
@dataclass(frozen=True, kw_only=True)
@ -172,7 +173,7 @@ async def async_setup_entry(
platform = async_get_current_platform()
platform.async_register_entity_service(
"ptz_move",
SERVICE_PTZ_MOVE,
{vol.Required(ATTR_SPEED): cv.positive_int},
"async_ptz_move",
[SUPPORT_PTZ_SPEED],

View File

@ -6,8 +6,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from reolink_aio.api import Chime
from homeassistant.components.reolink import const
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
from homeassistant.components.reolink.const import CONF_USE_HTTPS, DOMAIN
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
@ -137,14 +137,14 @@ def reolink_platforms() -> Generator[None]:
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Add the reolink mock config entry to hass."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: DEFAULT_PROTOCOL,

View File

@ -4,7 +4,8 @@ from unittest.mock import MagicMock, patch
from freezegun.api import FrozenDateTimeFactory
from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL, const
from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL
from homeassistant.components.reolink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant
@ -45,7 +46,7 @@ async def test_motion_sensor(
# test webhook callback
reolink_connect.motion_detected.return_value = True
reolink_connect.ONVIF_event_callback.return_value = [0]
webhook_id = f"{const.DOMAIN}_{TEST_UID.replace(':', '')}_ONVIF"
webhook_id = f"{DOMAIN}_{TEST_UID.replace(':', '')}_ONVIF"
client = await hass_client_no_auth()
await client.post(f"/api/webhook/{webhook_id}", data="test_data")

View File

@ -0,0 +1,112 @@
"""Test the Reolink button platform."""
from unittest.mock import MagicMock, patch
import pytest
from reolink_aio.exceptions import ReolinkError
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, SERVICE_PRESS
from homeassistant.components.reolink.button import ATTR_SPEED, SERVICE_PTZ_MOVE
from homeassistant.components.reolink.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from .conftest import TEST_NVR_NAME
from tests.common import MockConfigEntry
async def test_button(
hass: HomeAssistant,
config_entry: MockConfigEntry,
reolink_connect: MagicMock,
) -> None:
"""Test button entity with ptz up."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BUTTON]):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
entity_id = f"{Platform.BUTTON}.{TEST_NVR_NAME}_ptz_up"
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
reolink_connect.set_ptz_command.assert_called_once()
reolink_connect.set_ptz_command.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
async def test_ptz_move_service(
hass: HomeAssistant,
config_entry: MockConfigEntry,
reolink_connect: MagicMock,
) -> None:
"""Test ptz_move entity service using PTZ button entity."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BUTTON]):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
entity_id = f"{Platform.BUTTON}.{TEST_NVR_NAME}_ptz_up"
await hass.services.async_call(
DOMAIN,
SERVICE_PTZ_MOVE,
{ATTR_ENTITY_ID: entity_id, ATTR_SPEED: 5},
blocking=True,
)
reolink_connect.set_ptz_command.assert_called_with(0, command="Up", speed=5)
reolink_connect.set_ptz_command.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
DOMAIN,
SERVICE_PTZ_MOVE,
{ATTR_ENTITY_ID: entity_id, ATTR_SPEED: 5},
blocking=True,
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_host_button(
hass: HomeAssistant,
config_entry: MockConfigEntry,
reolink_connect: MagicMock,
) -> None:
"""Test host button entity with reboot."""
with patch("homeassistant.components.reolink.PLATFORMS", [Platform.BUTTON]):
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
entity_id = f"{Platform.BUTTON}.{TEST_NVR_NAME}_restart"
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
reolink_connect.reboot.assert_called_once()
reolink_connect.reboot.side_effect = ReolinkError("Test error")
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
BUTTON_DOMAIN,
SERVICE_PRESS,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)

View File

@ -10,8 +10,9 @@ from reolink_aio.exceptions import ApiError, CredentialsInvalidError, ReolinkErr
from homeassistant import config_entries
from homeassistant.components import dhcp
from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL, const
from homeassistant.components.reolink import DEVICE_UPDATE_INTERVAL
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
from homeassistant.components.reolink.const import CONF_USE_HTTPS, DOMAIN
from homeassistant.components.reolink.exceptions import ReolinkWebhookException
from homeassistant.components.reolink.host import DEFAULT_TIMEOUT
from homeassistant.config_entries import ConfigEntryState
@ -50,7 +51,7 @@ async def test_config_flow_manual_success(
) -> None:
"""Successful flow manually initialized by the user."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
@ -73,7 +74,7 @@ async def test_config_flow_manual_success(
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
}
assert result["options"] == {
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -85,7 +86,7 @@ async def test_config_flow_errors(
) -> None:
"""Successful flow manually initialized by the user after some errors."""
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
@ -206,7 +207,7 @@ async def test_config_flow_errors(
CONF_PASSWORD: TEST_PASSWORD,
CONF_HOST: TEST_HOST,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
)
@ -217,7 +218,7 @@ async def test_config_flow_errors(
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
}
assert result["options"] == {
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -227,14 +228,14 @@ async def test_config_flow_errors(
async def test_options_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
"""Test specifying non default settings using options flow."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: "rtsp",
@ -267,14 +268,14 @@ async def test_change_connection_settings(
) -> None:
"""Test changing connection settings by issuing a second user config flow."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -284,7 +285,7 @@ async def test_change_connection_settings(
config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_USER}
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
@ -310,14 +311,14 @@ async def test_change_connection_settings(
async def test_reauth(hass: HomeAssistant, mock_setup_entry: MagicMock) -> None:
"""Test a reauth flow."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -367,7 +368,7 @@ async def test_dhcp_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> No
)
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
)
assert result["type"] is FlowResultType.FORM
@ -389,7 +390,7 @@ async def test_dhcp_flow(hass: HomeAssistant, mock_setup_entry: MagicMock) -> No
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
}
assert result["options"] == {
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -442,14 +443,14 @@ async def test_dhcp_ip_update(
) -> None:
"""Test dhcp discovery aborts if already configured where the IP is updated if appropriate."""
config_entry = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC),
data={
CONF_HOST: TEST_HOST,
CONF_USERNAME: TEST_USERNAME,
CONF_PASSWORD: TEST_PASSWORD,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: DEFAULT_PROTOCOL,
@ -479,7 +480,7 @@ async def test_dhcp_ip_update(
setattr(reolink_connect, attr, value)
result = await hass.config_entries.flow.async_init(
const.DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
DOMAIN, context={"source": config_entries.SOURCE_DHCP}, data=dhcp_data
)
for host in host_call_list:

View File

@ -13,8 +13,8 @@ from homeassistant.components.reolink import (
DEVICE_UPDATE_INTERVAL,
FIRMWARE_UPDATE_INTERVAL,
NUM_CRED_ERRORS,
const,
)
from homeassistant.components.reolink.const import DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import STATE_OFF, STATE_UNAVAILABLE, Platform
@ -140,7 +140,7 @@ async def test_credential_error_three(
reolink_connect.get_states.side_effect = CredentialsInvalidError("Test error")
issue_id = f"config_entry_reauth_{const.DOMAIN}_{config_entry.entry_id}"
issue_id = f"config_entry_reauth_{DOMAIN}_{config_entry.entry_id}"
for _ in range(NUM_CRED_ERRORS):
assert (HOMEASSISTANT_DOMAIN, issue_id) not in issue_registry.issues
freezer.tick(DEVICE_UPDATE_INTERVAL)
@ -414,14 +414,14 @@ async def test_migrate_entity_ids(
reolink_connect.supported = mock_supported
dev_entry = device_registry.async_get_or_create(
identifiers={(const.DOMAIN, original_dev_id)},
identifiers={(DOMAIN, original_dev_id)},
config_entry_id=config_entry.entry_id,
disabled_by=None,
)
entity_registry.async_get_or_create(
domain=domain,
platform=const.DOMAIN,
platform=DOMAIN,
unique_id=original_id,
config_entry=config_entry,
suggested_object_id=original_id,
@ -429,16 +429,13 @@ async def test_migrate_entity_ids(
device_id=dev_entry.id,
)
assert entity_registry.async_get_entity_id(domain, const.DOMAIN, original_id)
assert entity_registry.async_get_entity_id(domain, const.DOMAIN, new_id) is None
assert entity_registry.async_get_entity_id(domain, DOMAIN, original_id)
assert entity_registry.async_get_entity_id(domain, DOMAIN, new_id) is None
assert device_registry.async_get_device(
identifiers={(const.DOMAIN, original_dev_id)}
)
assert device_registry.async_get_device(identifiers={(DOMAIN, original_dev_id)})
if new_dev_id != original_dev_id:
assert (
device_registry.async_get_device(identifiers={(const.DOMAIN, new_dev_id)})
is None
device_registry.async_get_device(identifiers={(DOMAIN, new_dev_id)}) is None
)
# setup CH 0 and host entities/device
@ -446,19 +443,15 @@ async def test_migrate_entity_ids(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (
entity_registry.async_get_entity_id(domain, const.DOMAIN, original_id) is None
)
assert entity_registry.async_get_entity_id(domain, const.DOMAIN, new_id)
assert entity_registry.async_get_entity_id(domain, DOMAIN, original_id) is None
assert entity_registry.async_get_entity_id(domain, DOMAIN, new_id)
if new_dev_id != original_dev_id:
assert (
device_registry.async_get_device(
identifiers={(const.DOMAIN, original_dev_id)}
)
device_registry.async_get_device(identifiers={(DOMAIN, original_dev_id)})
is None
)
assert device_registry.async_get_device(identifiers={(const.DOMAIN, new_dev_id)})
assert device_registry.async_get_device(identifiers={(DOMAIN, new_dev_id)})
async def test_no_repair_issue(
@ -472,11 +465,11 @@ async def test_no_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "https_webhook") not in issue_registry.issues
assert (const.DOMAIN, "webhook_url") not in issue_registry.issues
assert (const.DOMAIN, "enable_port") not in issue_registry.issues
assert (const.DOMAIN, "firmware_update") not in issue_registry.issues
assert (const.DOMAIN, "ssl") not in issue_registry.issues
assert (DOMAIN, "https_webhook") not in issue_registry.issues
assert (DOMAIN, "webhook_url") not in issue_registry.issues
assert (DOMAIN, "enable_port") not in issue_registry.issues
assert (DOMAIN, "firmware_update") not in issue_registry.issues
assert (DOMAIN, "ssl") not in issue_registry.issues
async def test_https_repair_issue(
@ -503,7 +496,7 @@ async def test_https_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "https_webhook") in issue_registry.issues
assert (DOMAIN, "https_webhook") in issue_registry.issues
async def test_ssl_repair_issue(
@ -533,7 +526,7 @@ async def test_ssl_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "ssl") in issue_registry.issues
assert (DOMAIN, "ssl") in issue_registry.issues
@pytest.mark.parametrize("protocol", ["rtsp", "rtmp"])
@ -553,7 +546,7 @@ async def test_port_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "enable_port") in issue_registry.issues
assert (DOMAIN, "enable_port") in issue_registry.issues
async def test_webhook_repair_issue(
@ -576,7 +569,7 @@ async def test_webhook_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "webhook_url") in issue_registry.issues
assert (DOMAIN, "webhook_url") in issue_registry.issues
async def test_firmware_repair_issue(
@ -590,4 +583,4 @@ async def test_firmware_repair_issue(
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert (const.DOMAIN, "firmware_update_host") in issue_registry.issues
assert (DOMAIN, "firmware_update_host") in issue_registry.issues

View File

@ -14,9 +14,8 @@ from homeassistant.components.media_source import (
async_resolve_media,
)
from homeassistant.components.media_source.error import Unresolvable
from homeassistant.components.reolink import const
from homeassistant.components.reolink.config_flow import DEFAULT_PROTOCOL
from homeassistant.components.reolink.const import DOMAIN
from homeassistant.components.reolink.const import CONF_USE_HTTPS, DOMAIN
from homeassistant.components.stream import DOMAIN as MEDIA_STREAM_DOMAIN
from homeassistant.const import (
CONF_HOST,
@ -321,14 +320,14 @@ async def test_browsing_not_loaded(
reolink_connect.get_host_data.side_effect = ReolinkError("Test error")
config_entry2 = MockConfigEntry(
domain=const.DOMAIN,
domain=DOMAIN,
unique_id=format_mac(TEST_MAC2),
data={
CONF_HOST: TEST_HOST2,
CONF_USERNAME: TEST_USERNAME2,
CONF_PASSWORD: TEST_PASSWORD2,
CONF_PORT: TEST_PORT,
const.CONF_USE_HTTPS: TEST_USE_HTTPS,
CONF_USE_HTTPS: TEST_USE_HTTPS,
},
options={
CONF_PROTOCOL: DEFAULT_PROTOCOL,