Add error handling for NVR event fetching in Hikvision integration (#160251)

Co-authored-by: Paul Tarjan <ptarjan@users.noreply.github.com>
This commit is contained in:
Paul Tarjan
2026-01-22 06:37:47 -07:00
committed by GitHub
parent 947ed121dc
commit fa30ed1dd8
3 changed files with 87 additions and 2 deletions

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
import logging
from xml.etree.ElementTree import ParseError
from pyhik.constants import SENSOR_MAP
from pyhik.hikvision import HikCamera
@@ -88,7 +89,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) ->
def fetch_and_inject_nvr_events() -> None:
"""Fetch and inject NVR events in a single executor job."""
nvr_events = camera.get_event_triggers(nvr_notification_methods)
try:
nvr_events = camera.get_event_triggers(nvr_notification_methods)
except (requests.exceptions.RequestException, ParseError) as err:
_LOGGER.warning("Unable to fetch event triggers from %s: %s", host, err)
return
_LOGGER.debug("NVR events fetched with extended methods: %s", nvr_events)
if nvr_events:
# Map raw event type names to friendly names using SENSOR_MAP
@@ -101,6 +107,12 @@ async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) ->
mapped_events[friendly_name] = list(channels)
_LOGGER.debug("Mapped NVR events: %s", mapped_events)
camera.inject_events(mapped_events)
else:
_LOGGER.debug(
"No event triggers returned from %s. "
"Ensure events are configured on the device",
host,
)
await hass.async_add_executor_job(fetch_and_inject_nvr_events)

View File

@@ -150,7 +150,12 @@ async def async_setup_entry(
sensors = camera.current_event_states
if sensors is None or not sensors:
_LOGGER.warning("Hikvision device has no sensors available")
_LOGGER.warning(
"Hikvision %s %s has no sensors available. "
"Ensure event detection is enabled and configured on the device",
data.device_type,
data.device_name,
)
return
async_add_entities(

View File

@@ -1,7 +1,9 @@
"""Test Hikvision integration setup and unload."""
from unittest.mock import MagicMock
from xml.etree.ElementTree import ParseError
import pytest
import requests
from homeassistant.config_entries import ConfigEntryState
@@ -102,3 +104,69 @@ async def test_setup_entry_nvr_fetches_events(
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hik_nvr.return_value.get_event_triggers.assert_called_once()
mock_hik_nvr.return_value.inject_events.assert_called_once()
async def test_setup_entry_nvr_event_fetch_request_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hik_nvr: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test setup continues when NVR event fetch fails with request error."""
mock_hik_nvr.return_value.get_event_triggers.side_effect = (
requests.exceptions.RequestException("Connection error")
)
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hik_nvr.return_value.get_event_triggers.assert_called_once()
mock_hik_nvr.return_value.inject_events.assert_not_called()
assert f"Unable to fetch event triggers from {TEST_HOST}" in caplog.text
async def test_setup_entry_nvr_event_fetch_parse_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hik_nvr: MagicMock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test setup continues when NVR event fetch fails with parse error."""
mock_hik_nvr.return_value.get_event_triggers.side_effect = ParseError("Invalid XML")
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hik_nvr.return_value.get_event_triggers.assert_called_once()
mock_hik_nvr.return_value.inject_events.assert_not_called()
assert f"Unable to fetch event triggers from {TEST_HOST}" in caplog.text
async def test_setup_entry_nvr_no_events_returned(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hik_nvr: MagicMock,
) -> None:
"""Test setup continues when NVR returns no events."""
mock_hik_nvr.return_value.get_event_triggers.return_value = None
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hik_nvr.return_value.get_event_triggers.assert_called_once()
mock_hik_nvr.return_value.inject_events.assert_not_called()
async def test_setup_entry_nvr_empty_events_returned(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_hik_nvr: MagicMock,
) -> None:
"""Test setup continues when NVR returns empty events."""
mock_hik_nvr.return_value.get_event_triggers.return_value = {}
await setup_integration(hass, mock_config_entry)
assert mock_config_entry.state is ConfigEntryState.LOADED
mock_hik_nvr.return_value.get_event_triggers.assert_called_once()
mock_hik_nvr.return_value.inject_events.assert_not_called()