evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add motion detection support to motionEye (#49665)

Browse source
This commit is contained in:
Dermot Duffy 2021-07-01 02:22:43 -07:00 • committed by GitHub
parent 7bd8e2aa55
commit 2868fef7d4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 751 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

201
homeassistant/components/motioneye/__init__.py
View file

@ -2,19 +2,46 @@
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Callable
from urllib.parse import urlencode, urljoin
from aiohttp.web import Request, Response
from motioneye_client.client import (
MotionEyeClient,
MotionEyeClientError,
MotionEyeClientInvalidAuthError,
)
from motioneye_client.const import KEY_CAMERAS, KEY_ID, KEY_NAME
from motioneye_client.const import (
KEY_CAMERAS,
KEY_HTTP_METHOD_POST_JSON,
KEY_ID,
KEY_NAME,
KEY_WEB_HOOK_CONVERSION_SPECIFIERS,
KEY_WEB_HOOK_NOTIFICATIONS_ENABLED,
KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD,
KEY_WEB_HOOK_NOTIFICATIONS_URL,
KEY_WEB_HOOK_STORAGE_ENABLED,
KEY_WEB_HOOK_STORAGE_HTTP_METHOD,
KEY_WEB_HOOK_STORAGE_URL,
)
from homeassistant.components.camera.const import DOMAIN as CAMERA_DOMAIN
from homeassistant.components.webhook import (
async_generate_id,
async_generate_path,
async_register as webhook_register,
async_unregister as webhook_unregister,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.const import (
ATTR_DEVICE_ID,
ATTR_NAME,
CONF_URL,
CONF_WEBHOOK_ID,
HTTP_BAD_REQUEST,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
@ -22,23 +49,35 @@ from homeassistant.helpers.dispatcher import (
async_dispatcher_connect,
async_dispatcher_send,
)
from homeassistant.helpers.network import get_url
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
ATTR_EVENT_TYPE,
ATTR_WEBHOOK_ID,
CONF_ADMIN_PASSWORD,
CONF_ADMIN_USERNAME,
CONF_CLIENT,
CONF_COORDINATOR,
CONF_SURVEILLANCE_PASSWORD,
CONF_SURVEILLANCE_USERNAME,
CONF_WEBHOOK_SET,
CONF_WEBHOOK_SET_OVERWRITE,
DEFAULT_SCAN_INTERVAL,
DEFAULT_WEBHOOK_SET,
DEFAULT_WEBHOOK_SET_OVERWRITE,
DOMAIN,
EVENT_FILE_STORED,
EVENT_FILE_STORED_KEYS,
EVENT_MOTION_DETECTED,
EVENT_MOTION_DETECTED_KEYS,
MOTIONEYE_MANUFACTURER,
SIGNAL_CAMERA_ADD,
WEB_HOOK_SENTINEL_KEY,
WEB_HOOK_SENTINEL_VALUE,
)
_LOGGER = logging.getLogger(__name__)
PLATFORMS = [CAMERA_DOMAIN]
@ -97,6 +136,15 @@ def listen_for_new_cameras(
)
@callback
def async_generate_motioneye_webhook(hass: HomeAssistant, webhook_id: str) -> str:
"""Generate the full local URL for a webhook_id."""
return "{}{}".format(
get_url(hass, allow_cloud=False),
async_generate_path(webhook_id),
)
@callback
def _add_camera(
hass: HomeAssistant,
@ -109,13 +157,93 @@ def _add_camera(
) -> None:
"""Add a motionEye camera to hass."""
device_registry.async_get_or_create(
def _is_recognized_web_hook(url: str) -> bool:
"""Determine whether this integration set a web hook."""
return f"{WEB_HOOK_SENTINEL_KEY}={WEB_HOOK_SENTINEL_VALUE}" in url
def _set_webhook(
url: str,
key_url: str,
key_method: str,
key_enabled: str,
camera: dict[str, Any],
) -> bool:
"""Set a web hook."""
if (
entry.options.get(
CONF_WEBHOOK_SET_OVERWRITE,
DEFAULT_WEBHOOK_SET_OVERWRITE,
)
or not camera.get(key_url)
or _is_recognized_web_hook(camera[key_url])
) and (
not camera.get(key_enabled, False)
or camera.get(key_method) != KEY_HTTP_METHOD_POST_JSON
or camera.get(key_url) != url
):
camera[key_enabled] = True
camera[key_method] = KEY_HTTP_METHOD_POST_JSON
camera[key_url] = url
return True
return False
def _build_url(
device: dr.DeviceEntry, base: str, event_type: str, keys: list[str]
) -> str:
"""Build a motionEye webhook URL."""
# This URL-surgery cannot use YARL because the output must NOT be
# url-encoded. This is because motionEye will do further string
# manipulation/substitution on this value before ultimately fetching it,
# and it cannot deal with URL-encoded input to that string manipulation.
return urljoin(
base,
"?"
+ urlencode(
{
**{k: KEY_WEB_HOOK_CONVERSION_SPECIFIERS[k] for k in sorted(keys)},
WEB_HOOK_SENTINEL_KEY: WEB_HOOK_SENTINEL_VALUE,
ATTR_EVENT_TYPE: event_type,
ATTR_DEVICE_ID: device.id,
},
safe="%{}",
),
)
device = device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
identifiers={device_identifier},
manufacturer=MOTIONEYE_MANUFACTURER,
model=MOTIONEYE_MANUFACTURER,
name=camera[KEY_NAME],
)
if entry.options.get(CONF_WEBHOOK_SET, DEFAULT_WEBHOOK_SET):
url = async_generate_motioneye_webhook(hass, entry.data[CONF_WEBHOOK_ID])
if _set_webhook(
_build_url(
device,
url,
EVENT_MOTION_DETECTED,
EVENT_MOTION_DETECTED_KEYS,
),
KEY_WEB_HOOK_NOTIFICATIONS_URL,
KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD,
KEY_WEB_HOOK_NOTIFICATIONS_ENABLED,
camera,
) | _set_webhook(
_build_url(
device,
url,
EVENT_FILE_STORED,
EVENT_FILE_STORED_KEYS,
),
KEY_WEB_HOOK_STORAGE_URL,
KEY_WEB_HOOK_STORAGE_HTTP_METHOD,
KEY_WEB_HOOK_STORAGE_ENABLED,
camera,
):
hass.async_create_task(client.async_set_camera(camera_id, camera))
async_dispatcher_send(
hass,
@ -124,6 +252,11 @@ def _add_camera(
)
async def _async_entry_updated(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
"""Handle entry updates."""
await hass.config_entries.async_reload(config_entry.entry_id)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up motionEye from a config entry."""
hass.data.setdefault(DOMAIN, {})
@ -145,6 +278,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await client.async_client_close()
raise ConfigEntryNotReady from exc
# Ensure every loaded entry has a registered webhook id.
if CONF_WEBHOOK_ID not in entry.data:
hass.config_entries.async_update_entry(
entry, data={**entry.data, CONF_WEBHOOK_ID: async_generate_id()}
)
webhook_register(
hass, DOMAIN, "motionEye", entry.data[CONF_WEBHOOK_ID], handle_webhook
)
@callback
async def async_update_data() -> dict[str, Any] | None:
try:
@ -196,8 +338,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
device_identifier,
)
# Ensure every device associated with this config entry is still in the list of
# motionEye cameras, otherwise remove the device (and thus entities).
# Ensure every device associated with this config entry is still in the
# list of motionEye cameras, otherwise remove the device (and thus
# entities).
for device_entry in dr.async_entries_for_config_entry(
device_registry, entry.entry_id
):
@ -218,6 +361,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
coordinator.async_add_listener(_async_process_motioneye_cameras)
)
await coordinator.async_refresh()
entry.async_on_unload(entry.add_update_listener(_async_entry_updated))
hass.async_create_task(setup_then_listen())
return True
@ -225,9 +369,54 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Unload a config entry."""
webhook_unregister(hass, entry.data[CONF_WEBHOOK_ID])
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
config_data = hass.data[DOMAIN].pop(entry.entry_id)
await config_data[CONF_CLIENT].async_client_close()
return unload_ok
async def handle_webhook(
hass: HomeAssistant, webhook_id: str, request: Request
) -> None | Response:
"""Handle webhook callback."""
try:
data = await request.json()
except (json.decoder.JSONDecodeError, UnicodeDecodeError):
return Response(
text="Could not decode request",
status=HTTP_BAD_REQUEST,
)
for key in (ATTR_DEVICE_ID, ATTR_EVENT_TYPE):
if key not in data:
return Response(
text=f"Missing webhook parameter: {key}",
status=HTTP_BAD_REQUEST,
)
event_type = data[ATTR_EVENT_TYPE]
device_registry = dr.async_get(hass)
device_id = data[ATTR_DEVICE_ID]
device = device_registry.async_get(device_id)
if not device:
return Response(
text=f"Device not found: {device_id}",
status=HTTP_BAD_REQUEST,
)
hass.bus.async_fire(
f"{DOMAIN}.{event_type}",
{
ATTR_DEVICE_ID: device.id,
ATTR_NAME: device.name,
ATTR_WEBHOOK_ID: webhook_id,
**data,
},
)
return None

57
homeassistant/components/motioneye/config_flow.py
View file

@ -11,8 +11,14 @@ from motioneye_client.client import (
)
import voluptuous as vol
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow
from homeassistant.const import CONF_SOURCE, CONF_URL
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigFlow,
OptionsFlow,
)
from homeassistant.const import CONF_SOURCE, CONF_URL, CONF_WEBHOOK_ID
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
@ -22,6 +28,10 @@ from .const import (
CONF_ADMIN_USERNAME,
CONF_SURVEILLANCE_PASSWORD,
CONF_SURVEILLANCE_USERNAME,
CONF_WEBHOOK_SET,
CONF_WEBHOOK_SET_OVERWRITE,
DEFAULT_WEBHOOK_SET,
DEFAULT_WEBHOOK_SET_OVERWRITE,
DOMAIN,
)
@ -122,6 +132,9 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
return _get_form(user_input, errors)
if self.context.get(CONF_SOURCE) == SOURCE_REAUTH and reauth_entry is not None:
# Persist the same webhook id across reauths.
if CONF_WEBHOOK_ID in reauth_entry.data:
user_input[CONF_WEBHOOK_ID] = reauth_entry.data[CONF_WEBHOOK_ID]
self.hass.config_entries.async_update_entry(reauth_entry, data=user_input)
# Need to manually reload, as the listener won't have been
# installed because the initial load did not succeed (the reauth
@ -167,3 +180,43 @@ class MotionEyeConfigFlow(ConfigFlow, domain=DOMAIN):
)
return await self.async_step_user()
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> MotionEyeOptionsFlow:
"""Get the Hyperion Options flow."""
return MotionEyeOptionsFlow(config_entry)
class MotionEyeOptionsFlow(OptionsFlow):
"""motionEye options flow."""
def __init__(self, config_entry: ConfigEntry) -> None:
"""Initialize a motionEye options flow."""
self._config_entry = config_entry
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
schema: dict[vol.Marker, type] = {
vol.Required(
CONF_WEBHOOK_SET,
default=self._config_entry.options.get(
CONF_WEBHOOK_SET,
DEFAULT_WEBHOOK_SET,
),
): bool,
vol.Required(
CONF_WEBHOOK_SET_OVERWRITE,
default=self._config_entry.options.get(
CONF_WEBHOOK_SET_OVERWRITE,
DEFAULT_WEBHOOK_SET_OVERWRITE,
),
): bool,
}
return self.async_show_form(step_id="init", data_schema=vol.Schema(schema))

94
homeassistant/components/motioneye/const.py
View file

@ -1,19 +1,89 @@
"""Constants for the motionEye integration."""
from datetime import timedelta
from typing import Final
DOMAIN = "motioneye"
from motioneye_client.const import (
KEY_WEB_HOOK_CS_CAMERA_ID,
KEY_WEB_HOOK_CS_CHANGED_PIXELS,
KEY_WEB_HOOK_CS_DESPECKLE_LABELS,
KEY_WEB_HOOK_CS_EVENT,
KEY_WEB_HOOK_CS_FILE_PATH,
KEY_WEB_HOOK_CS_FILE_TYPE,
KEY_WEB_HOOK_CS_FPS,
KEY_WEB_HOOK_CS_FRAME_NUMBER,
KEY_WEB_HOOK_CS_HEIGHT,
KEY_WEB_HOOK_CS_HOST,
KEY_WEB_HOOK_CS_MOTION_CENTER_X,
KEY_WEB_HOOK_CS_MOTION_CENTER_Y,
KEY_WEB_HOOK_CS_MOTION_HEIGHT,
KEY_WEB_HOOK_CS_MOTION_VERSION,
KEY_WEB_HOOK_CS_MOTION_WIDTH,
KEY_WEB_HOOK_CS_NOISE_LEVEL,
KEY_WEB_HOOK_CS_THRESHOLD,
KEY_WEB_HOOK_CS_WIDTH,
)
CONF_CLIENT = "client"
CONF_COORDINATOR = "coordinator"
CONF_ADMIN_PASSWORD = "admin_password"
CONF_ADMIN_USERNAME = "admin_username"
CONF_SURVEILLANCE_USERNAME = "surveillance_username"
CONF_SURVEILLANCE_PASSWORD = "surveillance_password"
DEFAULT_SCAN_INTERVAL = timedelta(seconds=30)
DOMAIN: Final = "motioneye"
MOTIONEYE_MANUFACTURER = "motionEye"
ATTR_EVENT_TYPE: Final = "event_type"
ATTR_WEBHOOK_ID: Final = "webhook_id"
SIGNAL_CAMERA_ADD = f"{DOMAIN}_camera_add_signal." "{}"
SIGNAL_CAMERA_REMOVE = f"{DOMAIN}_camera_remove_signal." "{}"
CONF_CLIENT: Final = "client"
CONF_COORDINATOR: Final = "coordinator"
CONF_ADMIN_PASSWORD: Final = "admin_password"
CONF_ADMIN_USERNAME: Final = "admin_username"
CONF_SURVEILLANCE_USERNAME: Final = "surveillance_username"
CONF_SURVEILLANCE_PASSWORD: Final = "surveillance_password"
CONF_WEBHOOK_SET: Final = "webhook_set"
CONF_WEBHOOK_SET_OVERWRITE: Final = "webhook_set_overwrite"
TYPE_MOTIONEYE_MJPEG_CAMERA = "motioneye_mjpeg_camera"
DEFAULT_WEBHOOK_SET: Final = True
DEFAULT_WEBHOOK_SET_OVERWRITE: Final = False
DEFAULT_SCAN_INTERVAL: Final = timedelta(seconds=30)
EVENT_MOTION_DETECTED: Final = "motion_detected"
EVENT_FILE_STORED: Final = "file_stored"
EVENT_MOTION_DETECTED_KEYS: Final = [
KEY_WEB_HOOK_CS_EVENT,
KEY_WEB_HOOK_CS_FRAME_NUMBER,
KEY_WEB_HOOK_CS_CAMERA_ID,
KEY_WEB_HOOK_CS_CHANGED_PIXELS,
KEY_WEB_HOOK_CS_NOISE_LEVEL,
KEY_WEB_HOOK_CS_WIDTH,
KEY_WEB_HOOK_CS_HEIGHT,
KEY_WEB_HOOK_CS_MOTION_WIDTH,
KEY_WEB_HOOK_CS_MOTION_HEIGHT,
KEY_WEB_HOOK_CS_MOTION_CENTER_X,
KEY_WEB_HOOK_CS_MOTION_CENTER_Y,
KEY_WEB_HOOK_CS_THRESHOLD,
KEY_WEB_HOOK_CS_DESPECKLE_LABELS,
KEY_WEB_HOOK_CS_FPS,
KEY_WEB_HOOK_CS_HOST,
KEY_WEB_HOOK_CS_MOTION_VERSION,
]
EVENT_FILE_STORED_KEYS: Final = [
KEY_WEB_HOOK_CS_EVENT,
KEY_WEB_HOOK_CS_FRAME_NUMBER,
KEY_WEB_HOOK_CS_CAMERA_ID,
KEY_WEB_HOOK_CS_NOISE_LEVEL,
KEY_WEB_HOOK_CS_WIDTH,
KEY_WEB_HOOK_CS_HEIGHT,
KEY_WEB_HOOK_CS_FILE_PATH,
KEY_WEB_HOOK_CS_FILE_TYPE,
KEY_WEB_HOOK_CS_THRESHOLD,
KEY_WEB_HOOK_CS_FPS,
KEY_WEB_HOOK_CS_HOST,
KEY_WEB_HOOK_CS_MOTION_VERSION,
]
MOTIONEYE_MANUFACTURER: Final = "motionEye"
SIGNAL_CAMERA_ADD: Final = f"{DOMAIN}_camera_add_signal." "{}"
SIGNAL_CAMERA_REMOVE: Final = f"{DOMAIN}_camera_remove_signal." "{}"
TYPE_MOTIONEYE_MJPEG_CAMERA: Final = "motioneye_mjpeg_camera"
WEB_HOOK_SENTINEL_KEY: Final = "src"
WEB_HOOK_SENTINEL_VALUE: Final = "hass-motioneye"

6
homeassistant/components/motioneye/manifest.json
View file

@ -3,8 +3,12 @@
"name": "motionEye",
"documentation": "https://www.home-assistant.io/integrations/motioneye",
"config_flow": true,
"dependencies": [
"http",
"webhook"
],
"requirements": [
"motioneye-client==0.3.6"
"motioneye-client==0.3.9"
],
"codeowners": [
"@dermotduffy"

10
homeassistant/components/motioneye/strings.json
View file

@ -25,5 +25,15 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"options": {
"step": {
"init": {
"data": {
"webhook_set": "Configure motionEye webhooks to report events to Home Assistant",
"webhook_set_overwrite": "Overwrite unrecognized webhooks"
}
}
}
}
}

2
requirements_all.txt
View file

@ -985,7 +985,7 @@ mitemp_bt==0.0.3
motionblinds==0.4.10
# homeassistant.components.motioneye
motioneye-client==0.3.6
motioneye-client==0.3.9
# homeassistant.components.mullvad
mullvad-api==1.0.0

2
requirements_test_all.txt
View file

@ -554,7 +554,7 @@ minio==4.0.9
motionblinds==0.4.10
# homeassistant.components.motioneye
motioneye-client==0.3.6
motioneye-client==0.3.9
# homeassistant.components.mullvad
mullvad-api==1.0.0

13
tests/components/motioneye/__init__.py
View file

@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, Mock, patch
from motioneye_client.const import DEFAULT_PORT
from homeassistant.components.motioneye.const import DOMAIN
from homeassistant.config import async_process_ha_core_config
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
@ -151,14 +152,14 @@ def create_mock_motioneye_config_entry(
options: dict[str, Any] | None = None,
) -> ConfigEntry:
"""Add a test config entry."""
config_entry: MockConfigEntry = MockConfigEntry( # type: ignore[no-untyped-call]
config_entry: MockConfigEntry = MockConfigEntry(
entry_id=TEST_CONFIG_ENTRY_ID,
domain=DOMAIN,
data=data or {CONF_URL: TEST_URL},
title=f"{TEST_URL}",
options=options or {},
)
config_entry.add_to_hass(hass) # type: ignore[no-untyped-call]
config_entry.add_to_hass(hass)
return config_entry
@ -167,7 +168,13 @@ async def setup_mock_motioneye_config_entry(
config_entry: ConfigEntry | None = None,
client: Mock | None = None,
) -> ConfigEntry:
"""Add a mock MotionEye config entry to hass."""
"""Create and setup a mock motionEye config entry."""
await async_process_ha_core_config(
hass,
{"external_url": "https://example.com"},
)
config_entry = config_entry or create_mock_motioneye_config_entry(hass)
client = client or create_mock_motioneye_client()

8
tests/components/motioneye/test_camera.py
View file

@ -1,7 +1,7 @@
"""Test the motionEye camera."""
import copy
import logging
from typing import Any
from typing import Any, cast
from unittest.mock import AsyncMock, Mock
from aiohttp import web
@ -235,7 +235,7 @@ async def test_get_still_image_from_camera(
# It won't actually get a stream from the dummy handler, so just catch
# the expected exception, then verify the right handler was called.
with pytest.raises(HomeAssistantError):
await async_get_image(hass, TEST_CAMERA_ENTITY_ID, timeout=None) # type: ignore[no-untyped-call]
await async_get_image(hass, TEST_CAMERA_ENTITY_ID, timeout=1)
assert image_handler.called
@ -269,7 +269,9 @@ async def test_get_stream_from_camera(aiohttp_server: Any, hass: HomeAssistant)
# It won't actually get a stream from the dummy handler, so just catch
# the expected exception, then verify the right handler was called.
with pytest.raises(HTTPBadGateway):
await async_get_mjpeg_stream(hass, None, TEST_CAMERA_ENTITY_ID) # type: ignore[no-untyped-call]
await async_get_mjpeg_stream(
hass, cast(web.Request, None), TEST_CAMERA_ENTITY_ID
)
assert stream_handler.called

43
tests/components/motioneye/test_config_flow.py
View file

@ -14,9 +14,11 @@ from homeassistant.components.motioneye.const import (
CONF_ADMIN_USERNAME,
CONF_SURVEILLANCE_PASSWORD,
CONF_SURVEILLANCE_USERNAME,
CONF_WEBHOOK_SET,
CONF_WEBHOOK_SET_OVERWRITE,
DOMAIN,
)
from homeassistant.const import CONF_URL
from homeassistant.const import CONF_URL, CONF_WEBHOOK_ID
from homeassistant.core import HomeAssistant
from . import TEST_URL, create_mock_motioneye_client, create_mock_motioneye_config_entry
@ -247,6 +249,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
"""Test a reauth."""
config_data = {
CONF_URL: TEST_URL,
CONF_WEBHOOK_ID: "test-webhook-id",
}
config_entry = create_mock_motioneye_config_entry(hass, data=config_data)
@ -287,7 +290,7 @@ async def test_reauth(hass: HomeAssistant) -> None:
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "reauth_successful"
assert dict(config_entry.data) == new_data
assert dict(config_entry.data) == {**new_data, CONF_WEBHOOK_ID: "test-webhook-id"}
assert len(mock_setup_entry.mock_calls) == 1
assert mock_client.async_client_close.called
@ -300,11 +303,11 @@ async def test_duplicate(hass: HomeAssistant) -> None:
}
# Add an existing entry with the same URL.
existing_entry: MockConfigEntry = MockConfigEntry( # type: ignore[no-untyped-call]
existing_entry: MockConfigEntry = MockConfigEntry(
domain=DOMAIN,
data=config_data,
)
existing_entry.add_to_hass(hass) # type: ignore[no-untyped-call]
existing_entry.add_to_hass(hass)
# Now do the usual config entry process, and verify it is rejected.
create_mock_motioneye_config_entry(hass, data=config_data)
@ -431,3 +434,35 @@ async def test_hassio_clean_up_on_user_flow(hass: HomeAssistant) -> None:
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 0
async def test_options(hass: HomeAssistant) -> None:
"""Check an options flow."""
config_entry = create_mock_motioneye_config_entry(hass)
client = create_mock_motioneye_client()
with patch(
"homeassistant.components.motioneye.MotionEyeClient",
return_value=client,
), patch(
"homeassistant.components.motioneye.async_setup_entry",
return_value=True,
):
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(config_entry.entry_id)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "init"
result = await hass.config_entries.options.async_configure(
result["flow_id"],
user_input={
CONF_WEBHOOK_SET: True,
CONF_WEBHOOK_SET_OVERWRITE: True,
},
)
await hass.async_block_till_done()
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_WEBHOOK_SET]
assert result["data"][CONF_WEBHOOK_SET_OVERWRITE]

348
tests/components/motioneye/test_web_hooks.py Normal file
View file

@ -0,0 +1,348 @@
"""Test the motionEye camera web hooks."""
import copy
import logging
from typing import Any
from unittest.mock import AsyncMock, call, patch
from motioneye_client.const import (
KEY_CAMERAS,
KEY_HTTP_METHOD_POST_JSON,
KEY_WEB_HOOK_NOTIFICATIONS_ENABLED,
KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD,
KEY_WEB_HOOK_NOTIFICATIONS_URL,
KEY_WEB_HOOK_STORAGE_ENABLED,
KEY_WEB_HOOK_STORAGE_HTTP_METHOD,
KEY_WEB_HOOK_STORAGE_URL,
)
from homeassistant.components.motioneye.const import (
ATTR_EVENT_TYPE,
CONF_WEBHOOK_SET_OVERWRITE,
DOMAIN,
EVENT_FILE_STORED,
EVENT_MOTION_DETECTED,
)
from homeassistant.components.webhook import URL_WEBHOOK_PATH
from homeassistant.const import (
ATTR_DEVICE_ID,
CONF_URL,
CONF_WEBHOOK_ID,
HTTP_BAD_REQUEST,
HTTP_OK,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.setup import async_setup_component
from . import (
TEST_CAMERA,
TEST_CAMERA_DEVICE_IDENTIFIER,
TEST_CAMERA_ID,
TEST_CAMERA_NAME,
TEST_CAMERAS,
TEST_URL,
create_mock_motioneye_client,
create_mock_motioneye_config_entry,
setup_mock_motioneye_config_entry,
)
from tests.common import async_capture_events
_LOGGER = logging.getLogger(__name__)
WEB_HOOK_MOTION_DETECTED_QUERY_STRING = (
"camera_id=%t&changed_pixels=%D&despeckle_labels=%Q&event=%v&fps=%{fps}"
"&frame_number=%q&height=%h&host=%{host}&motion_center_x=%K&motion_center_y=%L"
"&motion_height=%J&motion_version=%{ver}&motion_width=%i&noise_level=%N"
"&threshold=%o&width=%w&src=hass-motioneye&event_type=motion_detected"
)
WEB_HOOK_FILE_STORED_QUERY_STRING = (
"camera_id=%t&event=%v&file_path=%f&file_type=%n&fps=%{fps}&frame_number=%q"
"&height=%h&host=%{host}&motion_version=%{ver}&noise_level=%N&threshold=%o&width=%w"
"&src=hass-motioneye&event_type=file_stored"
)
async def test_setup_camera_without_webhook(hass: HomeAssistant) -> None:
"""Test a camera with no webhook."""
client = create_mock_motioneye_client()
config_entry = await setup_mock_motioneye_config_entry(hass, client=client)
device_registry = await dr.async_get_registry(hass)
device = device_registry.async_get_device(
identifiers={TEST_CAMERA_DEVICE_IDENTIFIER}
)
assert device
expected_camera = copy.deepcopy(TEST_CAMERA)
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_ENABLED] = True
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_MOTION_DETECTED_QUERY_STRING}&device_id={device.id}"
)
expected_camera[KEY_WEB_HOOK_STORAGE_ENABLED] = True
expected_camera[KEY_WEB_HOOK_STORAGE_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_STORAGE_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_FILE_STORED_QUERY_STRING}&device_id={device.id}"
)
assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
async def test_setup_camera_with_wrong_webhook(
hass: HomeAssistant,
) -> None:
"""Test camera with wrong web hook."""
wrong_url = "http://wrong-url"
client = create_mock_motioneye_client()
cameras = copy.deepcopy(TEST_CAMERAS)
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_NOTIFICATIONS_URL] = wrong_url
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_STORAGE_URL] = wrong_url
client.async_get_cameras = AsyncMock(return_value=cameras)
config_entry = create_mock_motioneye_config_entry(hass)
await setup_mock_motioneye_config_entry(
hass,
config_entry=config_entry,
client=client,
)
assert not client.async_set_camera.called
# Update the options, which will trigger a reload with the new behavior.
with patch(
"homeassistant.components.motioneye.MotionEyeClient",
return_value=client,
):
hass.config_entries.async_update_entry(
config_entry, options={CONF_WEBHOOK_SET_OVERWRITE: True}
)
await hass.async_block_till_done()
device_registry = await dr.async_get_registry(hass)
device = device_registry.async_get_device(
identifiers={TEST_CAMERA_DEVICE_IDENTIFIER}
)
assert device
expected_camera = copy.deepcopy(TEST_CAMERA)
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_ENABLED] = True
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_MOTION_DETECTED_QUERY_STRING}&device_id={device.id}"
)
expected_camera[KEY_WEB_HOOK_STORAGE_ENABLED] = True
expected_camera[KEY_WEB_HOOK_STORAGE_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_STORAGE_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_FILE_STORED_QUERY_STRING}&device_id={device.id}"
)
assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
async def test_setup_camera_with_old_webhook(
hass: HomeAssistant,
) -> None:
"""Verify that webhooks are overwritten if they are from this integration.
Even if the overwrite option is disabled, verify the behavior is still to
overwrite incorrect versions of the URL that were set by this integration.
(To allow the web hook URL to be seamlessly updated in future versions)
"""
old_url = "http://old-url?src=hass-motioneye"
client = create_mock_motioneye_client()
cameras = copy.deepcopy(TEST_CAMERAS)
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_NOTIFICATIONS_URL] = old_url
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_STORAGE_URL] = old_url
client.async_get_cameras = AsyncMock(return_value=cameras)
config_entry = create_mock_motioneye_config_entry(hass)
await setup_mock_motioneye_config_entry(
hass,
config_entry=config_entry,
client=client,
)
assert client.async_set_camera.called
device_registry = await dr.async_get_registry(hass)
device = device_registry.async_get_device(
identifiers={TEST_CAMERA_DEVICE_IDENTIFIER}
)
assert device
expected_camera = copy.deepcopy(TEST_CAMERA)
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_ENABLED] = True
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_NOTIFICATIONS_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_MOTION_DETECTED_QUERY_STRING}&device_id={device.id}"
)
expected_camera[KEY_WEB_HOOK_STORAGE_ENABLED] = True
expected_camera[KEY_WEB_HOOK_STORAGE_HTTP_METHOD] = KEY_HTTP_METHOD_POST_JSON
expected_camera[KEY_WEB_HOOK_STORAGE_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_FILE_STORED_QUERY_STRING}&device_id={device.id}"
)
assert client.async_set_camera.call_args == call(TEST_CAMERA_ID, expected_camera)
async def test_setup_camera_with_correct_webhook(
hass: HomeAssistant,
) -> None:
"""Verify that webhooks are not overwritten if they are already correct."""
client = create_mock_motioneye_client()
config_entry = create_mock_motioneye_config_entry(
hass, data={CONF_URL: TEST_URL, CONF_WEBHOOK_ID: "webhook_secret_id"}
)
device_registry = await dr.async_get_registry(hass)
device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={TEST_CAMERA_DEVICE_IDENTIFIER},
)
cameras = copy.deepcopy(TEST_CAMERAS)
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_NOTIFICATIONS_ENABLED] = True
cameras[KEY_CAMERAS][0][
KEY_WEB_HOOK_NOTIFICATIONS_HTTP_METHOD
] = KEY_HTTP_METHOD_POST_JSON
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_NOTIFICATIONS_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_MOTION_DETECTED_QUERY_STRING}&device_id={device.id}"
)
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_STORAGE_ENABLED] = True
cameras[KEY_CAMERAS][0][
KEY_WEB_HOOK_STORAGE_HTTP_METHOD
] = KEY_HTTP_METHOD_POST_JSON
cameras[KEY_CAMERAS][0][KEY_WEB_HOOK_STORAGE_URL] = (
"https://example.com"
+ URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID])
+ f"?{WEB_HOOK_FILE_STORED_QUERY_STRING}&device_id={device.id}"
)
client.async_get_cameras = AsyncMock(return_value=cameras)
await setup_mock_motioneye_config_entry(
hass,
config_entry=config_entry,
client=client,
)
# Webhooks are correctly configured, so no set call should have been made.
assert not client.async_set_camera.called
async def test_good_query(hass: HomeAssistant, aiohttp_client: Any) -> None:
"""Test good callbacks."""
await async_setup_component(hass, "http", {"http": {}})
device_registry = await dr.async_get_registry(hass)
client = create_mock_motioneye_client()
config_entry = await setup_mock_motioneye_config_entry(hass, client=client)
device = device_registry.async_get_or_create(
config_entry_id=config_entry.entry_id,
identifiers={TEST_CAMERA_DEVICE_IDENTIFIER},
)
data = {
"one": "1",
"two": "2",
ATTR_DEVICE_ID: device.id,
}
client = await aiohttp_client(hass.http.app)
for event in (EVENT_MOTION_DETECTED, EVENT_FILE_STORED):
events = async_capture_events(hass, f"{DOMAIN}.{event}")
resp = await client.post(
URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID]),
json={
**data,
ATTR_EVENT_TYPE: event,
},
)
assert resp.status == HTTP_OK
assert len(events) == 1
assert events[0].data == {
"name": TEST_CAMERA_NAME,
"device_id": device.id,
ATTR_EVENT_TYPE: event,
CONF_WEBHOOK_ID: config_entry.data[CONF_WEBHOOK_ID],
**data,
}
async def test_bad_query_missing_parameters(
hass: HomeAssistant, aiohttp_client: Any
) -> None:
"""Test a query with missing parameters."""
await async_setup_component(hass, "http", {"http": {}})
config_entry = await setup_mock_motioneye_config_entry(hass)
client = await aiohttp_client(hass.http.app)
resp = await client.post(
URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID]), json={}
)
assert resp.status == HTTP_BAD_REQUEST
async def test_bad_query_no_such_device(
hass: HomeAssistant, aiohttp_client: Any
) -> None:
"""Test a correct query with incorrect device."""
await async_setup_component(hass, "http", {"http": {}})
config_entry = await setup_mock_motioneye_config_entry(hass)
client = await aiohttp_client(hass.http.app)
resp = await client.post(
URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID]),
json={
ATTR_EVENT_TYPE: EVENT_MOTION_DETECTED,
ATTR_DEVICE_ID: "not-a-real-device",
},
)
assert resp.status == HTTP_BAD_REQUEST
async def test_bad_query_cannot_decode(
hass: HomeAssistant, aiohttp_client: Any
) -> None:
"""Test a correct query with incorrect device."""
await async_setup_component(hass, "http", {"http": {}})
config_entry = await setup_mock_motioneye_config_entry(hass)
client = await aiohttp_client(hass.http.app)
motion_events = async_capture_events(hass, f"{DOMAIN}.{EVENT_MOTION_DETECTED}")
storage_events = async_capture_events(hass, f"{DOMAIN}.{EVENT_FILE_STORED}")
resp = await client.post(
URL_WEBHOOK_PATH.format(webhook_id=config_entry.data[CONF_WEBHOOK_ID]),
data=b"this is not json",
)
assert resp.status == HTTP_BAD_REQUEST
assert not motion_events
assert not storage_events