Add a reboot button for ONVIF devices (#61522)
This commit is contained in:
Eric Severance
GitHub
parent
a046cef734
commit
5f2fd1b0e6
5 changed files with 245 additions and 127 deletions
|
|
@ -83,7 +83,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
|||
|
||||
hass.data[DOMAIN][entry.unique_id] = device
|
||||
|
||||
platforms = [Platform.CAMERA]
|
||||
platforms = [Platform.BUTTON, Platform.CAMERA]
|
||||
|
||||
if device.capabilities.events:
|
||||
platforms += [Platform.BINARY_SENSOR, Platform.SENSOR]
|
||||
|
|
|
|||
41
homeassistant/components/onvif/button.py
Normal file
41
homeassistant/components/onvif/button.py
Normal file
|
|
@ -0,0 +1,41 @@
|
|||
"""ONVIF Buttons."""
|
||||
|
||||
from homeassistant.components.button import ButtonDeviceClass, ButtonEntity
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ENTITY_CATEGORY_CONFIG
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from .base import ONVIFBaseEntity
|
||||
from .const import DOMAIN
|
||||
from .device import ONVIFDevice
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up ONVIF button based on a config entry."""
|
||||
device = hass.data[DOMAIN][config_entry.unique_id]
|
||||
async_add_entities([RebootButton(device)])
|
||||
|
||||
|
||||
class RebootButton(ONVIFBaseEntity, ButtonEntity):
|
||||
"""Defines a ONVIF reboot button."""
|
||||
|
||||
_attr_device_class = ButtonDeviceClass.RESTART
|
||||
_attr_entity_category = ENTITY_CATEGORY_CONFIG
|
||||
|
||||
def __init__(self, device: ONVIFDevice) -> None:
|
||||
"""Initialize the button entity."""
|
||||
super().__init__(device)
|
||||
self._attr_name = f"{self.device.name} Reboot"
|
||||
self._attr_unique_id = (
|
||||
f"{self.device.info.mac or self.device.info.serial_number}_reboot"
|
||||
)
|
||||
|
||||
async def async_press(self) -> None:
|
||||
"""Send out a SystemReboot command."""
|
||||
device_mgmt = self.device.device.create_devicemgmt_service()
|
||||
await device_mgmt.SystemReboot()
|
||||
|
|
@ -1 +1,149 @@
|
|||
"""Tests for the ONVIF integration."""
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from zeep.exceptions import Fault
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.onvif import config_flow
|
||||
from homeassistant.components.onvif.const import CONF_SNAPSHOT_AUTH
|
||||
from homeassistant.components.onvif.models import DeviceInfo
|
||||
from homeassistant.const import HTTP_DIGEST_AUTHENTICATION
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
URN = "urn:uuid:123456789"
|
||||
NAME = "TestCamera"
|
||||
HOST = "1.2.3.4"
|
||||
PORT = 80
|
||||
USERNAME = "admin"
|
||||
PASSWORD = "12345"
|
||||
MAC = "aa:bb:cc:dd:ee"
|
||||
SERIAL_NUMBER = "ABCDEFGHIJK"
|
||||
MANUFACTURER = "TestManufacturer"
|
||||
MODEL = "TestModel"
|
||||
FIRMWARE_VERSION = "TestFirmwareVersion"
|
||||
|
||||
|
||||
def setup_mock_onvif_camera(
|
||||
mock_onvif_camera,
|
||||
with_h264=True,
|
||||
two_profiles=False,
|
||||
with_interfaces=True,
|
||||
with_interfaces_not_implemented=False,
|
||||
with_serial=True,
|
||||
):
|
||||
"""Prepare mock onvif.ONVIFCamera."""
|
||||
devicemgmt = MagicMock()
|
||||
|
||||
device_info = MagicMock()
|
||||
device_info.SerialNumber = SERIAL_NUMBER if with_serial else None
|
||||
devicemgmt.GetDeviceInformation = AsyncMock(return_value=device_info)
|
||||
|
||||
interface = MagicMock()
|
||||
interface.Enabled = True
|
||||
interface.Info.HwAddress = MAC
|
||||
|
||||
if with_interfaces_not_implemented:
|
||||
devicemgmt.GetNetworkInterfaces = AsyncMock(
|
||||
side_effect=Fault("not implemented")
|
||||
)
|
||||
else:
|
||||
devicemgmt.GetNetworkInterfaces = AsyncMock(
|
||||
return_value=[interface] if with_interfaces else []
|
||||
)
|
||||
|
||||
media_service = MagicMock()
|
||||
|
||||
profile1 = MagicMock()
|
||||
profile1.VideoEncoderConfiguration.Encoding = "H264" if with_h264 else "MJPEG"
|
||||
profile2 = MagicMock()
|
||||
profile2.VideoEncoderConfiguration.Encoding = "H264" if two_profiles else "MJPEG"
|
||||
|
||||
media_service.GetProfiles = AsyncMock(return_value=[profile1, profile2])
|
||||
|
||||
mock_onvif_camera.update_xaddrs = AsyncMock(return_value=True)
|
||||
mock_onvif_camera.create_devicemgmt_service = MagicMock(return_value=devicemgmt)
|
||||
mock_onvif_camera.create_media_service = MagicMock(return_value=media_service)
|
||||
mock_onvif_camera.close = AsyncMock(return_value=None)
|
||||
|
||||
def mock_constructor(
|
||||
host,
|
||||
port,
|
||||
user,
|
||||
passwd,
|
||||
wsdl_dir,
|
||||
encrypt=True,
|
||||
no_cache=False,
|
||||
adjust_time=False,
|
||||
transport=None,
|
||||
):
|
||||
"""Fake the controller constructor."""
|
||||
return mock_onvif_camera
|
||||
|
||||
mock_onvif_camera.side_effect = mock_constructor
|
||||
|
||||
|
||||
def setup_mock_device(mock_device):
|
||||
"""Prepare mock ONVIFDevice."""
|
||||
mock_device.async_setup = AsyncMock(return_value=True)
|
||||
mock_device.available = True
|
||||
mock_device.name = NAME
|
||||
mock_device.info = DeviceInfo(
|
||||
MANUFACTURER,
|
||||
MODEL,
|
||||
FIRMWARE_VERSION,
|
||||
SERIAL_NUMBER,
|
||||
MAC,
|
||||
)
|
||||
|
||||
def mock_constructor(hass, config):
|
||||
"""Fake the controller constructor."""
|
||||
return mock_device
|
||||
|
||||
mock_device.side_effect = mock_constructor
|
||||
|
||||
|
||||
async def setup_onvif_integration(
|
||||
hass,
|
||||
config=None,
|
||||
options=None,
|
||||
unique_id=MAC,
|
||||
entry_id="1",
|
||||
source=config_entries.SOURCE_USER,
|
||||
):
|
||||
"""Create an ONVIF config entry."""
|
||||
if not config:
|
||||
config = {
|
||||
config_flow.CONF_NAME: NAME,
|
||||
config_flow.CONF_HOST: HOST,
|
||||
config_flow.CONF_PORT: PORT,
|
||||
config_flow.CONF_USERNAME: USERNAME,
|
||||
config_flow.CONF_PASSWORD: PASSWORD,
|
||||
CONF_SNAPSHOT_AUTH: HTTP_DIGEST_AUTHENTICATION,
|
||||
}
|
||||
|
||||
config_entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
source=source,
|
||||
data={**config},
|
||||
options=options or {},
|
||||
entry_id=entry_id,
|
||||
unique_id=unique_id,
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onvif.config_flow.get_device"
|
||||
) as mock_onvif_camera, patch(
|
||||
"homeassistant.components.onvif.config_flow.wsdiscovery"
|
||||
) as mock_discovery, patch(
|
||||
"homeassistant.components.onvif.ONVIFDevice"
|
||||
) as mock_device:
|
||||
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
||||
# no discovery
|
||||
mock_discovery.return_value = []
|
||||
setup_mock_device(mock_device)
|
||||
mock_device.device = mock_onvif_camera
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
return config_entry, mock_onvif_camera, mock_device
|
||||
|
|
|
|||
40
tests/components/onvif/test_button.py
Normal file
40
tests/components/onvif/test_button.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
"""Test button of ONVIF integration."""
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN, ButtonDeviceClass
|
||||
from homeassistant.const import ATTR_DEVICE_CLASS, ATTR_ENTITY_ID, STATE_UNKNOWN
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import MAC, setup_onvif_integration
|
||||
|
||||
|
||||
async def test_reboot_button(hass):
|
||||
"""Test states of the Reboot button."""
|
||||
await setup_onvif_integration(hass)
|
||||
|
||||
state = hass.states.get("button.testcamera_reboot")
|
||||
assert state
|
||||
assert state.state == STATE_UNKNOWN
|
||||
assert state.attributes.get(ATTR_DEVICE_CLASS) == ButtonDeviceClass.RESTART
|
||||
|
||||
registry = er.async_get(hass)
|
||||
entry = registry.async_get("button.testcamera_reboot")
|
||||
assert entry
|
||||
assert entry.unique_id == f"{MAC}_reboot"
|
||||
|
||||
|
||||
async def test_reboot_button_press(hass):
|
||||
"""Test Reboot button press."""
|
||||
_, camera, _ = await setup_onvif_integration(hass)
|
||||
devicemgmt = camera.create_devicemgmt_service()
|
||||
devicemgmt.SystemReboot = AsyncMock(return_value=True)
|
||||
|
||||
await hass.services.async_call(
|
||||
BUTTON_DOMAIN,
|
||||
"press",
|
||||
{ATTR_ENTITY_ID: "button.testcamera_reboot"},
|
||||
blocking=True,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
devicemgmt.SystemReboot.assert_called_once()
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
"""Test ONVIF config flow."""
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from onvif.exceptions import ONVIFError
|
||||
from zeep.exceptions import Fault
|
||||
|
|
@ -7,16 +7,19 @@ from zeep.exceptions import Fault
|
|||
from homeassistant import config_entries, data_entry_flow
|
||||
from homeassistant.components.onvif import config_flow
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
URN = "urn:uuid:123456789"
|
||||
NAME = "TestCamera"
|
||||
HOST = "1.2.3.4"
|
||||
PORT = 80
|
||||
USERNAME = "admin"
|
||||
PASSWORD = "12345"
|
||||
MAC = "aa:bb:cc:dd:ee"
|
||||
SERIAL_NUMBER = "ABCDEFGHIJK"
|
||||
from . import (
|
||||
HOST,
|
||||
MAC,
|
||||
NAME,
|
||||
PASSWORD,
|
||||
PORT,
|
||||
SERIAL_NUMBER,
|
||||
URN,
|
||||
USERNAME,
|
||||
setup_mock_device,
|
||||
setup_mock_onvif_camera,
|
||||
setup_onvif_integration,
|
||||
)
|
||||
|
||||
DISCOVERY = [
|
||||
{
|
||||
|
|
@ -36,65 +39,6 @@ DISCOVERY = [
|
|||
]
|
||||
|
||||
|
||||
def setup_mock_onvif_camera(
|
||||
mock_onvif_camera,
|
||||
with_h264=True,
|
||||
two_profiles=False,
|
||||
with_interfaces=True,
|
||||
with_interfaces_not_implemented=False,
|
||||
with_serial=True,
|
||||
):
|
||||
"""Prepare mock onvif.ONVIFCamera."""
|
||||
devicemgmt = MagicMock()
|
||||
|
||||
device_info = MagicMock()
|
||||
device_info.SerialNumber = SERIAL_NUMBER if with_serial else None
|
||||
devicemgmt.GetDeviceInformation = AsyncMock(return_value=device_info)
|
||||
|
||||
interface = MagicMock()
|
||||
interface.Enabled = True
|
||||
interface.Info.HwAddress = MAC
|
||||
|
||||
if with_interfaces_not_implemented:
|
||||
devicemgmt.GetNetworkInterfaces = AsyncMock(
|
||||
side_effect=Fault("not implemented")
|
||||
)
|
||||
else:
|
||||
devicemgmt.GetNetworkInterfaces = AsyncMock(
|
||||
return_value=[interface] if with_interfaces else []
|
||||
)
|
||||
|
||||
media_service = MagicMock()
|
||||
|
||||
profile1 = MagicMock()
|
||||
profile1.VideoEncoderConfiguration.Encoding = "H264" if with_h264 else "MJPEG"
|
||||
profile2 = MagicMock()
|
||||
profile2.VideoEncoderConfiguration.Encoding = "H264" if two_profiles else "MJPEG"
|
||||
|
||||
media_service.GetProfiles = AsyncMock(return_value=[profile1, profile2])
|
||||
|
||||
mock_onvif_camera.update_xaddrs = AsyncMock(return_value=True)
|
||||
mock_onvif_camera.create_devicemgmt_service = MagicMock(return_value=devicemgmt)
|
||||
mock_onvif_camera.create_media_service = MagicMock(return_value=media_service)
|
||||
mock_onvif_camera.close = AsyncMock(return_value=None)
|
||||
|
||||
def mock_constructor(
|
||||
host,
|
||||
port,
|
||||
user,
|
||||
passwd,
|
||||
wsdl_dir,
|
||||
encrypt=True,
|
||||
no_cache=False,
|
||||
adjust_time=False,
|
||||
transport=None,
|
||||
):
|
||||
"""Fake the controller constructor."""
|
||||
return mock_onvif_camera
|
||||
|
||||
mock_onvif_camera.side_effect = mock_constructor
|
||||
|
||||
|
||||
def setup_mock_discovery(
|
||||
mock_discovery, with_name=False, with_mac=False, two_devices=False
|
||||
):
|
||||
|
|
@ -126,61 +70,6 @@ def setup_mock_discovery(
|
|||
mock_discovery.return_value = services
|
||||
|
||||
|
||||
def setup_mock_device(mock_device):
|
||||
"""Prepare mock ONVIFDevice."""
|
||||
mock_device.async_setup = AsyncMock(return_value=True)
|
||||
|
||||
def mock_constructor(hass, config):
|
||||
"""Fake the controller constructor."""
|
||||
return mock_device
|
||||
|
||||
mock_device.side_effect = mock_constructor
|
||||
|
||||
|
||||
async def setup_onvif_integration(
|
||||
hass,
|
||||
config=None,
|
||||
options=None,
|
||||
unique_id=MAC,
|
||||
entry_id="1",
|
||||
source=config_entries.SOURCE_USER,
|
||||
):
|
||||
"""Create an ONVIF config entry."""
|
||||
if not config:
|
||||
config = {
|
||||
config_flow.CONF_NAME: NAME,
|
||||
config_flow.CONF_HOST: HOST,
|
||||
config_flow.CONF_PORT: PORT,
|
||||
config_flow.CONF_USERNAME: USERNAME,
|
||||
config_flow.CONF_PASSWORD: PASSWORD,
|
||||
}
|
||||
|
||||
config_entry = MockConfigEntry(
|
||||
domain=config_flow.DOMAIN,
|
||||
source=source,
|
||||
data={**config},
|
||||
options=options or {},
|
||||
entry_id=entry_id,
|
||||
unique_id=unique_id,
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.onvif.config_flow.get_device"
|
||||
) as mock_onvif_camera, patch(
|
||||
"homeassistant.components.onvif.config_flow.wsdiscovery"
|
||||
) as mock_discovery, patch(
|
||||
"homeassistant.components.onvif.ONVIFDevice"
|
||||
) as mock_device:
|
||||
setup_mock_onvif_camera(mock_onvif_camera, two_profiles=True)
|
||||
# no discovery
|
||||
mock_discovery.return_value = []
|
||||
setup_mock_device(mock_device)
|
||||
await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
return config_entry
|
||||
|
||||
|
||||
async def test_flow_discovered_devices(hass):
|
||||
"""Test that config flow works for discovered devices."""
|
||||
|
||||
|
|
@ -616,7 +505,7 @@ async def test_flow_import_onvif_auth_error(hass):
|
|||
|
||||
async def test_option_flow(hass):
|
||||
"""Test config flow options."""
|
||||
entry = await setup_onvif_integration(hass)
|
||||
entry, _, _ = await setup_onvif_integration(hass)
|
||||
|
||||
result = await hass.config_entries.options.async_init(entry.entry_id)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue