core/tests/components/onvif/test_parsers.py

716 lines
26 KiB
Python

"""Test ONVIF parsers."""
import datetime
import os
import onvif
import onvif.settings
from zeep import Client
from zeep.transports import Transport
from homeassistant.components.onvif import models, parsers
from homeassistant.core import HomeAssistant
TEST_UID = "test-unique-id"
async def get_event(notification_data: dict) -> models.Event:
"""Take in a zeep dict, run it through the parser, and return an Event.
When the parser encounters an unknown topic that it doesn't know how to parse,
it outputs a message 'No registered handler for event from ...' along with a
print out of the serialized xml message from zeep. If it tries to parse and
can't, it prints out 'Unable to parse event from ...' along with the same
serialized message. This method can take the output directly from these log
messages and run them through the parser, which makes it easy to add new unit
tests that verify the message can now be parsed.
"""
zeep_client = Client(
f"{os.path.dirname(onvif.__file__)}/wsdl/events.wsdl",
wsse=None,
transport=Transport(),
)
notif_msg_type = zeep_client.get_type("ns5:NotificationMessageHolderType")
assert notif_msg_type is not None
notif_msg = notif_msg_type(**notification_data)
assert notif_msg is not None
# The xsd:any type embedded inside the message doesn't parse, so parse it manually.
msg_elem = zeep_client.get_element("ns8:Message")
assert msg_elem is not None
msg_data = msg_elem(**notification_data["Message"]["_value_1"])
assert msg_data is not None
notif_msg.Message._value_1 = msg_data
parser = parsers.PARSERS.get(notif_msg.Topic._value_1)
assert parser is not None
return await parser(TEST_UID, notif_msg)
async def test_line_detector_crossed(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/LineDetector/Crossed."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {"_value_1": None, "_attr_1": None},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/LineDetector/Crossed",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "xx.xx.xx.xx/onvif/event/alarm",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "video_source_config1",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "analytics_video_source",
},
{"Name": "Rule", "Value": "MyLineDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "ObjectId", "Value": "0"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(2020, 5, 24, 7, 24, 47),
"PropertyOperation": "Initialized",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Line Detector Crossed"
assert event.platform == "sensor"
assert event.value == "0"
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/LineDetector/"
"Crossed_video_source_config1_analytics_video_source_MyLineDetectorRule"
)
async def test_tapo_line_crossed(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/CellMotionDetector/LineCross."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/CellMotionDetector/LineCross",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:5656/event",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyLineCrossDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "IsLineCross", "Value": "true"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(
2025, 1, 3, 21, 5, 14, tzinfo=datetime.UTC
),
"PropertyOperation": "Changed",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Line Detector Crossed"
assert event.platform == "binary_sensor"
assert event.device_class == "motion"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/"
"LineCross_VideoSourceToken_VideoAnalyticsToken_MyLineCrossDetectorRule"
)
async def test_tapo_tpsmartevent_vehicle(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - vehicle."""
event = await get_event(
{
"Message": {
"_value_1": {
"Data": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [{"Name": "IsVehicle", "Value": "true"}],
"_attr_1": None,
},
"Extension": None,
"Key": None,
"PropertyOperation": "Changed",
"Source": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{
"Name": "Rule",
"Value": "MyTPSmartEventDetectorRule",
},
],
"_attr_1": None,
},
"UtcTime": datetime.datetime(
2024, 11, 2, 0, 33, 11, tzinfo=datetime.UTC
),
"_attr_1": {},
}
},
"ProducerReference": {
"Address": {
"_attr_1": None,
"_value_1": "http://192.168.56.127:5656/event",
},
"Metadata": None,
"ReferenceParameters": None,
"_attr_1": None,
"_value_1": None,
},
"SubscriptionReference": {
"Address": {
"_attr_1": None,
"_value_1": "http://192.168.56.127:2020/event-0_2020",
},
"Metadata": None,
"ReferenceParameters": None,
"_attr_1": None,
"_value_1": None,
},
"Topic": {
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
"_value_1": "tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent",
},
}
)
assert event is not None
assert event.name == "Vehicle Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "motion"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/TPSmartEventDetector/"
"TPSmartEvent_VideoSourceToken_VideoAnalyticsToken_MyTPSmartEventDetectorRule"
)
async def test_tapo_cellmotiondetector_vehicle(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/CellMotionDetector/TpSmartEvent - vehicle."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/CellMotionDetector/TpSmartEvent",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:5656/event",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyTPSmartEventDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "IsVehicle", "Value": "true"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(
2025, 1, 5, 14, 2, 9, tzinfo=datetime.UTC
),
"PropertyOperation": "Changed",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Vehicle Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "motion"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/"
"TpSmartEvent_VideoSourceToken_VideoAnalyticsToken_MyTPSmartEventDetectorRule"
)
async def test_tapo_tpsmartevent_person(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/TPSmartEventDetector/TPSmartEvent - person."""
event = await get_event(
{
"Message": {
"_value_1": {
"Data": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [{"Name": "IsPeople", "Value": "true"}],
"_attr_1": None,
},
"Extension": None,
"Key": None,
"PropertyOperation": "Changed",
"Source": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyPeopleDetectorRule"},
],
"_attr_1": None,
},
"UtcTime": datetime.datetime(
2024, 11, 3, 18, 40, 43, tzinfo=datetime.UTC
),
"_attr_1": {},
}
},
"ProducerReference": {
"Address": {
"_attr_1": None,
"_value_1": "http://192.168.56.127:5656/event",
},
"Metadata": None,
"ReferenceParameters": None,
"_attr_1": None,
"_value_1": None,
},
"SubscriptionReference": {
"Address": {
"_attr_1": None,
"_value_1": "http://192.168.56.127:2020/event-0_2020",
},
"Metadata": None,
"ReferenceParameters": None,
"_attr_1": None,
"_value_1": None,
},
"Topic": {
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
"_value_1": "tns1:RuleEngine/PeopleDetector/People",
},
}
)
assert event is not None
assert event.name == "Person Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "motion"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/PeopleDetector/"
"People_VideoSourceToken_VideoAnalyticsToken_MyPeopleDetectorRule"
)
async def test_tapo_cellmotiondetector_person(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/CellMotionDetector/People - person."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {
"_value_1": "http://192.168.56.63:2020/event-0_2020",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/CellMotionDetector/People",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "http://192.168.56.63:5656/event",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyPeopleDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "IsPeople", "Value": "true"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(
2025, 1, 3, 20, 9, 22, tzinfo=datetime.UTC
),
"PropertyOperation": "Changed",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Person Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "motion"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/"
"People_VideoSourceToken_VideoAnalyticsToken_MyPeopleDetectorRule"
)
async def test_tapo_tamper(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/CellMotionDetector/Tamper - tamper."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:2020/event-0_2020",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/CellMotionDetector/Tamper",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "http://CAMERA_LOCAL_IP:5656/event",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyTamperDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "IsTamper", "Value": "true"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(
2025, 1, 5, 21, 1, 5, tzinfo=datetime.UTC
),
"PropertyOperation": "Changed",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Tamper Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "tamper"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/"
"Tamper_VideoSourceToken_VideoAnalyticsToken_MyTamperDetectorRule"
)
async def test_tapo_intrusion(hass: HomeAssistant) -> None:
"""Tests tns1:RuleEngine/CellMotionDetector/Intrusion - intrusion."""
event = await get_event(
{
"SubscriptionReference": {
"Address": {
"_value_1": "http://192.168.100.155:2020/event-0_2020",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Topic": {
"_value_1": "tns1:RuleEngine/CellMotionDetector/Intrusion",
"Dialect": "http://www.onvif.org/ver10/tev/topicExpression/ConcreteSet",
"_attr_1": {},
},
"ProducerReference": {
"Address": {
"_value_1": "http://192.168.100.155:5656/event",
"_attr_1": None,
},
"ReferenceParameters": None,
"Metadata": None,
"_value_1": None,
"_attr_1": None,
},
"Message": {
"_value_1": {
"Source": {
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyIntrusionDetectorRule"},
],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Key": None,
"Data": {
"SimpleItem": [{"Name": "IsIntrusion", "Value": "true"}],
"ElementItem": [],
"Extension": None,
"_attr_1": None,
},
"Extension": None,
"UtcTime": datetime.datetime(
2025, 1, 11, 10, 40, 45, tzinfo=datetime.UTC
),
"PropertyOperation": "Changed",
"_attr_1": {},
}
},
}
)
assert event is not None
assert event.name == "Intrusion Detection"
assert event.platform == "binary_sensor"
assert event.device_class == "safety"
assert event.value
assert event.uid == (
f"{TEST_UID}_tns1:RuleEngine/CellMotionDetector/"
"Intrusion_VideoSourceToken_VideoAnalyticsToken_MyIntrusionDetectorRule"
)
async def test_tapo_missing_attributes(hass: HomeAssistant) -> None:
"""Tests async_parse_tplink_detector with missing fields."""
event = await get_event(
{
"Message": {
"_value_1": {
"Data": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [{"Name": "IsPeople", "Value": "true"}],
"_attr_1": None,
},
}
},
"Topic": {
"_value_1": "tns1:RuleEngine/PeopleDetector/People",
},
}
)
assert event is None
async def test_tapo_unknown_type(hass: HomeAssistant) -> None:
"""Tests async_parse_tplink_detector with unknown event type."""
event = await get_event(
{
"Message": {
"_value_1": {
"Data": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [{"Name": "IsNotPerson", "Value": "true"}],
"_attr_1": None,
},
"Source": {
"ElementItem": [],
"Extension": None,
"SimpleItem": [
{
"Name": "VideoSourceConfigurationToken",
"Value": "vsconf",
},
{
"Name": "VideoAnalyticsConfigurationToken",
"Value": "VideoAnalyticsToken",
},
{"Name": "Rule", "Value": "MyPeopleDetectorRule"},
],
},
}
},
"Topic": {
"_value_1": "tns1:RuleEngine/PeopleDetector/People",
},
}
)
assert event is None