Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-04-18 07:56:03 +01:00
Code Activity

Update onvif parsers library to latest parsing multiple (#165571)

Browse Source 
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Jeff Terrace
2026-03-16 16:40:37 -04:00
committed by GitHub
parent 41c497c49e
commit 66f04c702c
6 changed files with 89 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
homeassistant/components/onvif/event.py
View File

@@ -17,6 +17,7 @@ from onvif.client import (
from onvif.exceptions import ONVIFError from onvif.exceptions import ONVIFError
from onvif.util import stringify_onvif_error from onvif.util import stringify_onvif_error
import onvif_parsers import onvif_parsers
import onvif_parsers.util
from zeep.exceptions import Fault, TransportError, ValidationError, XMLParseError from zeep.exceptions import Fault, TransportError, ValidationError, XMLParseError
from homeassistant.components import webhook from homeassistant.components import webhook
@@ -196,7 +197,7 @@ class EventManager:
topic = msg.Topic._value_1.rstrip("/.") # noqa: SLF001 topic = msg.Topic._value_1.rstrip("/.") # noqa: SLF001
try: try:
event = await onvif_parsers.parse(topic, unique_id, msg) events = await onvif_parsers.parse(topic, unique_id, msg)
error = None error = None
except onvif_parsers.errors.UnknownTopicError: except onvif_parsers.errors.UnknownTopicError:
if topic not in UNHANDLED_TOPICS: if topic not in UNHANDLED_TOPICS:
@@ -204,42 +205,43 @@ class EventManager:
"%s: No registered handler for event from %s: %s", "%s: No registered handler for event from %s: %s",
self.name, self.name,
unique_id, unique_id,
msg, onvif_parsers.util.event_to_debug_format(msg),
) )
UNHANDLED_TOPICS.add(topic) UNHANDLED_TOPICS.add(topic)
continue continue
except (AttributeError, KeyError) as e: except (AttributeError, KeyError) as e:
event = None events = []
error = e error = e
if not event: if not events:
LOGGER.warning( LOGGER.warning(
"%s: Unable to parse event from %s: %s: %s", "%s: Unable to parse event from %s: %s: %s",
self.name, self.name,
unique_id, unique_id,
error, error,
msg, onvif_parsers.util.event_to_debug_format(msg),
) )
continue continue
value = event.value for event in events:
if event.device_class == "timestamp" and isinstance(value, str): value = event.value
value = _local_datetime_or_none(value) if event.device_class == "timestamp" and isinstance(value, str):
value = _local_datetime_or_none(value)
ha_event = Event( ha_event = Event(
uid=event.uid, uid=event.uid,
name=event.name, name=event.name,
platform=event.platform, platform=event.platform,
device_class=event.device_class, device_class=event.device_class,
unit_of_measurement=event.unit_of_measurement, unit_of_measurement=event.unit_of_measurement,
value=value, value=value,
entity_category=ENTITY_CATEGORY_MAPPING.get( entity_category=ENTITY_CATEGORY_MAPPING.get(
event.entity_category or "" event.entity_category or ""
), ),
entity_enabled=event.entity_enabled, entity_enabled=event.entity_enabled,
) )
self.get_uids_by_platform(ha_event.platform).add(ha_event.uid) self.get_uids_by_platform(ha_event.platform).add(ha_event.uid)
self._events[ha_event.uid] = ha_event self._events[ha_event.uid] = ha_event
def get_uid(self, uid: str) -> Event | None: def get_uid(self, uid: str) -> Event | None:
"""Retrieve event for given id.""" """Retrieve event for given id."""

2
homeassistant/components/onvif/manifest.json
View File

@@ -15,7 +15,7 @@
"loggers": ["onvif", "wsdiscovery", "zeep"], "loggers": ["onvif", "wsdiscovery", "zeep"],
"requirements": [ "requirements": [
"onvif-zeep-async==4.0.4", "onvif-zeep-async==4.0.4",
"onvif_parsers==1.2.2", "onvif_parsers==2.3.0",
"WSDiscovery==2.1.2" "WSDiscovery==2.1.2"
] ]
} }

2
requirements_all.txt generated
View File

@@ -1688,7 +1688,7 @@ onedrive-personal-sdk==0.1.7
onvif-zeep-async==4.0.4 onvif-zeep-async==4.0.4
# homeassistant.components.onvif # homeassistant.components.onvif
onvif_parsers==1.2.2 onvif_parsers==2.3.0
# homeassistant.components.opengarage # homeassistant.components.opengarage
open-garage==0.2.0 open-garage==0.2.0

2
requirements_test_all.txt generated
View File

@@ -1474,7 +1474,7 @@ onedrive-personal-sdk==0.1.7
onvif-zeep-async==4.0.4 onvif-zeep-async==4.0.4
# homeassistant.components.onvif # homeassistant.components.onvif
onvif_parsers==1.2.2 onvif_parsers==2.3.0
# homeassistant.components.opengarage # homeassistant.components.opengarage
open-garage==0.2.0 open-garage==0.2.0

11
tests/components/onvif/__init__.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import collections
from collections import defaultdict from collections import defaultdict
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
@@ -196,7 +197,7 @@ async def setup_onvif_integration(
source=config_entries.SOURCE_USER, source=config_entries.SOURCE_USER,
capabilities=None, capabilities=None,
events=None, events=None,
raw_events: list[tuple[str, EventEntity]] | None = None, raw_events: list[tuple[str, list[EventEntity]]] | None = None,
) -> tuple[MockConfigEntry, MagicMock, MagicMock]: ) -> tuple[MockConfigEntry, MagicMock, MagicMock]:
"""Create an ONVIF config entry.""" """Create an ONVIF config entry."""
if not config: if not config:
@@ -239,12 +240,14 @@ async def setup_onvif_integration(
# to test the full parsing pipeline including conversions # to test the full parsing pipeline including conversions
event_manager = EventManager(hass, mock_onvif_camera, config_entry, NAME) event_manager = EventManager(hass, mock_onvif_camera, config_entry, NAME)
mock_messages = [] mock_messages = []
event_by_topic: dict[str, EventEntity] = {} event_by_topic: collections.defaultdict[str, list[EventEntity]] = (
for topic, raw_event in raw_events: collections.defaultdict(list)
)
for topic, topic_events in raw_events:
mock_msg = MagicMock() mock_msg = MagicMock()
mock_msg.Topic._value_1 = topic mock_msg.Topic._value_1 = topic
mock_messages.append(mock_msg) mock_messages.append(mock_msg)
event_by_topic[topic] = raw_event event_by_topic[topic].extend(topic_events)
async def mock_parse(topic, unique_id, msg): async def mock_parse(topic, unique_id, msg):
return event_by_topic.get(topic) return event_by_topic.get(topic)

69
tests/components/onvif/test_event.py
View File

@@ -95,13 +95,15 @@ async def test_timestamp_event_conversion(hass: HomeAssistant) -> None:
raw_events=[ raw_events=[
( (
"tns1:Monitoring/LastReset", "tns1:Monitoring/LastReset",
EventEntity( [
uid=LAST_RESET_UID, EventEntity(
name="Last Reset", uid=LAST_RESET_UID,
platform="sensor", name="Last Reset",
device_class="timestamp", platform="sensor",
value="2023-10-01T12:00:00Z", device_class="timestamp",
), value="2023-10-01T12:00:00Z",
),
],
), ),
], ],
) )
@@ -121,13 +123,15 @@ async def test_timestamp_event_invalid_value(hass: HomeAssistant) -> None:
raw_events=[ raw_events=[
( (
"tns1:Monitoring/LastReset", "tns1:Monitoring/LastReset",
EventEntity( [
uid=LAST_RESET_UID, EventEntity(
name="Last Reset", uid=LAST_RESET_UID,
platform="sensor", name="Last Reset",
device_class="timestamp", platform="sensor",
value="0000-00-00T00:00:00Z", device_class="timestamp",
), value="0000-00-00T00:00:00Z",
),
],
), ),
], ],
) )
@@ -135,3 +139,40 @@ async def test_timestamp_event_invalid_value(hass: HomeAssistant) -> None:
state = hass.states.get("sensor.testcamera_last_reset") state = hass.states.get("sensor.testcamera_last_reset")
assert state is not None assert state is not None
assert state.state == "unknown" assert state.state == "unknown"
async def test_multiple_events_same_topic(hass: HomeAssistant) -> None:
"""Test that multiple events with the same topic are all processed."""
await setup_onvif_integration(
hass,
capabilities=Capabilities(events=True, imaging=True, ptz=True),
raw_events=[
(
"tns1:VideoSource/MotionAlarm",
[
EventEntity(
uid=f"{MOTION_ALARM_UID}_1",
name="Motion Alarm 1",
platform="binary_sensor",
device_class="motion",
value=True,
),
EventEntity(
uid=f"{MOTION_ALARM_UID}_2",
name="Motion Alarm 2",
platform="binary_sensor",
device_class="motion",
value=False,
),
],
),
],
)
state1 = hass.states.get("binary_sensor.testcamera_motion_alarm_1")
assert state1 is not None
assert state1.state == STATE_ON
state2 = hass.states.get("binary_sensor.testcamera_motion_alarm_2")
assert state2 is not None
assert state2.state == STATE_OFF