core/tests/components/onvif/__init__.py

"""Tests for the ONVIF integration."""

from __future__ import annotations

from collections import defaultdict
from unittest.mock import AsyncMock, MagicMock, patch

from onvif.exceptions import ONVIFError
from onvif_parsers.model import EventEntity
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.event import EventManager
from homeassistant.components.onvif.models import (
    Capabilities,
    DeviceInfo,
    Event,
    Profile,
    PullPointManagerState,
    Resolution,
    Video,
    WebHookManagerState,
)
from homeassistant.const import HTTP_DIGEST_AUTHENTICATION
from homeassistant.core import HomeAssistant

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:ff"
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,
    profiles_transient_failure=False,
    auth_fail=False,
    update_xaddrs_fail=False,
    no_profiles=False,
    auth_failure=False,
    wrong_port=False,
):
    """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"

    if auth_fail:
        media_service.GetProfiles = AsyncMock(side_effect=Fault("Authority failure"))
    elif profiles_transient_failure:
        media_service.GetProfiles = AsyncMock(side_effect=Fault("camera not ready"))
    elif no_profiles:
        media_service.GetProfiles = AsyncMock(return_value=[])
    else:
        media_service.GetProfiles = AsyncMock(return_value=[profile1, profile2])

    if wrong_port:
        mock_onvif_camera.update_xaddrs = AsyncMock(side_effect=AttributeError)
    elif auth_failure:
        mock_onvif_camera.update_xaddrs = AsyncMock(
            side_effect=Fault(
                "not authorized", subcodes=[MagicMock(text="NotAuthorized")]
            )
        )
    elif update_xaddrs_fail:
        mock_onvif_camera.update_xaddrs = AsyncMock(
            side_effect=ONVIFError("camera not ready")
        )
    else:
        mock_onvif_camera.update_xaddrs = AsyncMock(return_value=True)
    mock_onvif_camera.create_devicemgmt_service = AsyncMock(return_value=devicemgmt)
    mock_onvif_camera.create_media_service = AsyncMock(return_value=media_service)
    mock_onvif_camera.close = AsyncMock(return_value=None)
    mock_onvif_camera.xaddrs = {}
    mock_onvif_camera.services = {}

    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, capabilities=None, profiles=None, events=None):
    """Prepare mock ONVIFDevice."""
    mock_device.async_setup = AsyncMock(return_value=True)
    mock_device.port = 80
    mock_device.available = True
    mock_device.name = NAME
    mock_device.info = DeviceInfo(
        MANUFACTURER,
        MODEL,
        FIRMWARE_VERSION,
        SERIAL_NUMBER,
        MAC,
    )
    mock_device.capabilities = capabilities or Capabilities(imaging=True, ptz=True)
    profile1 = Profile(
        index=0,
        token="dummy",
        name="profile1",
        video=Video("any", Resolution(640, 480)),
        ptz=None,
        video_source_token=None,
    )
    mock_device.profiles = profiles or [profile1]
    mock_device.events = MagicMock(
        webhook_manager=MagicMock(state=WebHookManagerState.STARTED),
        pullpoint_manager=MagicMock(state=PullPointManagerState.PAUSED),
        async_stop=AsyncMock(),
    )
    mock_device.device.close = AsyncMock()
    if events:
        _setup_mock_events(mock_device.events, events)

    def mock_constructor(
        hass: HomeAssistant, config: config_entries.ConfigEntry
    ) -> MagicMock:
        """Fake the controller constructor."""
        return mock_device

    mock_device.side_effect = mock_constructor


def _setup_mock_events(mock_events: MagicMock, events: list[Event]) -> None:
    """Configure mock events to return proper Event objects."""
    events_by_platform: dict[str, list[Event]] = defaultdict(list)
    events_by_uid: dict[str, Event] = {}
    uids_by_platform: dict[str, set[str]] = defaultdict(set)
    for event in events:
        events_by_platform[event.platform].append(event)
        events_by_uid[event.uid] = event
        uids_by_platform[event.platform].add(event.uid)

    mock_events.get_platform.side_effect = lambda p: list(events_by_platform.get(p, []))
    mock_events.get_uid.side_effect = events_by_uid.get
    mock_events.get_uids_by_platform.side_effect = lambda p: set(
        uids_by_platform.get(p, set())
    )


async def setup_onvif_integration(
    hass: HomeAssistant,
    config=None,
    options=None,
    unique_id=MAC,
    entry_id="1",
    source=config_entries.SOURCE_USER,
    capabilities=None,
    events=None,
    raw_events: list[tuple[str, EventEntity]] | None = None,
) -> tuple[MockConfigEntry, MagicMock, MagicMock]:
    """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, capabilities=capabilities, events=events)
        mock_device.device = mock_onvif_camera

        if raw_events:
            # Process raw library events through a real EventManager
            # to test the full parsing pipeline including conversions
            event_manager = EventManager(hass, mock_onvif_camera, config_entry, NAME)
            mock_messages = []
            event_by_topic: dict[str, EventEntity] = {}
            for topic, raw_event in raw_events:
                mock_msg = MagicMock()
                mock_msg.Topic._value_1 = topic
                mock_messages.append(mock_msg)
                event_by_topic[topic] = raw_event

            async def mock_parse(topic, unique_id, msg):
                return event_by_topic.get(topic)

            with patch(
                "homeassistant.components.onvif.event.onvif_parsers"
            ) as mock_parsers:
                mock_parsers.parse = mock_parse
                mock_parsers.errors.UnknownTopicError = type(
                    "UnknownTopicError", (Exception,), {}
                )
                await event_manager.async_parse_messages(mock_messages)

            mock_device.events = event_manager

        await hass.config_entries.async_setup(config_entry.entry_id)
        await hass.async_block_till_done()
    return config_entry, mock_onvif_camera, mock_device