diff --git a/homeassistant/components/vivotek/camera.py b/homeassistant/components/vivotek/camera.py index 6fde80808829..a726d3de6e6e 100644 --- a/homeassistant/components/vivotek/camera.py +++ b/homeassistant/components/vivotek/camera.py @@ -1,14 +1,17 @@ """Support for Vivotek IP Cameras.""" +from collections.abc import Callable +from functools import partial import logging -from typing import TYPE_CHECKING, Final, override +from typing import TYPE_CHECKING, Any, Final, override -from libpyvivotek.vivotek import VivotekCamera +from libpyvivotek.vivotek import VivotekCamera, VivotekCameraError from homeassistant.components.camera import Camera, CameraEntityFeature from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback from . import VivotekConfigEntry @@ -26,6 +29,40 @@ DEFAULT_STREAM_SOURCE = "live.sdp" PLATFORM_SCHEMA: Final = cv.removed(DOMAIN, raise_if_present=False) +def _fetch_str_metadata( + fetcher: Callable[[], Any], + log_message: str, +) -> str | None: + """Fetch optional string metadata from the camera.""" + try: + value: Any = fetcher() + except VivotekCameraError: + _LOGGER.debug(log_message) + return None + + return value if isinstance(value, str) else None + + +def _fetch_metadata( + cam_client: VivotekCamera, entry_title: str +) -> tuple[str | None, ...]: + """Fetch optional metadata from the camera in a single executor job.""" + return ( + _fetch_str_metadata( + cam_client.get_serial, + f"Failed to fetch serial number for {entry_title}", + ), + _fetch_str_metadata( + partial(cam_client.get_param, "system_info_firmwareversion"), + f"Failed to fetch firmware version for {entry_title}", + ), + _fetch_str_metadata( + partial(cam_client.get_param, "system_info_modelname"), + f"Failed to fetch model for {entry_title}", + ), + ) + + async def async_setup_entry( hass: HomeAssistant, entry: VivotekConfigEntry, @@ -38,6 +75,12 @@ async def async_setup_entry( f"rtsp://{creds}@{config[CONF_IP_ADDRESS]}:554/{config[CONF_STREAM_PATH]}" ) cam_client = entry.runtime_data + serial_number, sw_version, model = await hass.async_add_executor_job( + _fetch_metadata, + cam_client, + entry.title, + ) + if TYPE_CHECKING: assert entry.unique_id is not None async_add_entities( @@ -46,6 +89,9 @@ async def async_setup_entry( cam_client, stream_source, entry.unique_id, + serial_number, + sw_version, + model, entry.options[CONF_FRAMERATE], entry.title, ) @@ -64,6 +110,9 @@ class VivotekCam(Camera): cam_client: VivotekCamera, stream_source: str, unique_id: str, + serial_number: str | None, + sw_version: str | None, + model: str | None, framerate: int, name: str, ) -> None: @@ -73,7 +122,16 @@ class VivotekCam(Camera): self._attr_frame_interval = 1 / framerate self._attr_unique_id = unique_id self._attr_name = name + self._attr_available = True self._stream_source = stream_source + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, unique_id)}, + manufacturer=DEFAULT_CAMERA_BRAND, + model=model, + name=name, + serial_number=serial_number, + sw_version=sw_version, + ) @override def camera_image( @@ -101,5 +159,9 @@ class VivotekCam(Camera): def update(self) -> None: """Update entity status.""" - self._attr_model = self._cam.model_name - self._attr_available = self._attr_model is not None + try: + self._cam.get_serial() + except VivotekCameraError: + self._attr_available = False + else: + self._attr_available = True diff --git a/tests/components/vivotek/test_camera.py b/tests/components/vivotek/test_camera.py index 0c56129b68ad..b62aeee4047b 100644 --- a/tests/components/vivotek/test_camera.py +++ b/tests/components/vivotek/test_camera.py @@ -1,11 +1,23 @@ """Tests for the Vivotek camera integration.""" +from datetime import timedelta from unittest.mock import AsyncMock, patch +from freezegun.api import FrozenDateTimeFactory +from libpyvivotek.vivotek import VivotekCameraError from syrupy.assertion import SnapshotAssertion +from homeassistant.components.camera import ( + SERVICE_DISABLE_MOTION, + SERVICE_ENABLE_MOTION, + async_get_image, + async_get_stream_source, +) +from homeassistant.components.vivotek.const import DOMAIN +from homeassistant.const import STATE_UNAVAILABLE from homeassistant.core import HomeAssistant -from homeassistant.helpers import entity_registry as er +from homeassistant.helpers import device_registry as dr, entity_registry as er +from homeassistant.helpers.entity_component import async_update_entity from . import setup_integration @@ -24,3 +36,148 @@ async def test_all_entities( await setup_integration(hass, mock_config_entry) await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +async def test_camera_device_info( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, + device_registry: dr.DeviceRegistry, + entity_registry: er.EntityRegistry, +) -> None: + """Test the camera is linked to a device with expected metadata.""" + mock_vivotek_camera.get_serial.return_value = "ABCD1234" + mock_vivotek_camera.get_param.side_effect = lambda key: { + "system_info_firmwareversion": "1.2.3", + "system_info_modelname": "FD9165-HT", + }[key] + + await setup_integration(hass, mock_config_entry) + + entity_entry = entity_registry.async_get("camera.vivotek_camera") + assert entity_entry is not None + assert entity_entry.device_id is not None + + device = device_registry.async_get(entity_entry.device_id) + assert device is not None + assert (DOMAIN, "11:22:33:44:55:66") in device.identifiers + assert device.manufacturer == "VIVOTEK" + assert device.model == "FD9165-HT" + assert device.serial_number == "ABCD1234" + assert device.sw_version == "1.2.3" + + +async def test_camera_device_info_with_metadata_errors( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, + device_registry: dr.DeviceRegistry, + entity_registry: er.EntityRegistry, +) -> None: + """Test metadata fields are omitted when camera metadata calls fail.""" + mock_vivotek_camera.get_serial.side_effect = VivotekCameraError + mock_vivotek_camera.get_param.side_effect = VivotekCameraError + + await setup_integration(hass, mock_config_entry) + + entity_entry = entity_registry.async_get("camera.vivotek_camera") + assert entity_entry is not None + assert entity_entry.device_id is not None + + device = device_registry.async_get(entity_entry.device_id) + assert device is not None + assert device.model is None + assert device.serial_number is None + assert device.sw_version is None + + +async def test_camera_available_when_update_succeeds( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test camera is available when update probe succeeds.""" + await setup_integration(hass, mock_config_entry) + + freezer.tick(timedelta(seconds=1)) + await async_update_entity(hass, "camera.vivotek_camera") + + state = hass.states.get("camera.vivotek_camera") + assert state is not None + assert state.state != STATE_UNAVAILABLE + + +async def test_camera_unavailable_when_update_fails( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, + freezer: FrozenDateTimeFactory, +) -> None: + """Test camera is unavailable when update probe raises error.""" + await setup_integration(hass, mock_config_entry) + + mock_vivotek_camera.get_serial.side_effect = VivotekCameraError + freezer.tick(timedelta(seconds=1)) + await async_update_entity(hass, "camera.vivotek_camera") + + state = hass.states.get("camera.vivotek_camera") + assert state is not None + + assert state.state == STATE_UNAVAILABLE + + +async def test_camera_stream_source( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test stream source is returned from camera entity.""" + await setup_integration(hass, mock_config_entry) + + stream_source = await async_get_stream_source(hass, "camera.vivotek_camera") + + assert stream_source == "rtsp://admin:pass1234@1.2.3.4:554//live.sdp" + + +async def test_camera_motion_detection_methods( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test motion detection commands update entity state.""" + await setup_integration(hass, mock_config_entry) + + mock_vivotek_camera.set_param.side_effect = ["1", "0"] + + await hass.services.async_call( + "camera", + SERVICE_ENABLE_MOTION, + {"entity_id": "camera.vivotek_camera"}, + blocking=True, + ) + + await hass.services.async_call( + "camera", + SERVICE_DISABLE_MOTION, + {"entity_id": "camera.vivotek_camera"}, + blocking=True, + ) + + mock_vivotek_camera.set_param.assert_any_call("event_i0_enable", 1) + mock_vivotek_camera.set_param.assert_any_call("event_i0_enable", 0) + + +async def test_camera_image_returns_snapshot( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test camera image comes directly from camera snapshot call.""" + await setup_integration(hass, mock_config_entry) + + mock_vivotek_camera.snapshot.return_value = b"snapshot-bytes" + + image = await async_get_image(hass, "camera.vivotek_camera") + + assert image.content == b"snapshot-bytes"