From 4df5a41b5721979d951829233af1cf92691c13e5 Mon Sep 17 00:00:00 2001 From: Paul Tarjan Date: Tue, 16 Dec 2025 07:44:23 -0700 Subject: [PATCH] Migrate Hikvision integration to config flow (#158279) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Kamil BreguĊ‚a --- CODEOWNERS | 1 + .../components/hikvision/__init__.py | 88 ++++- .../components/hikvision/binary_sensor.py | 314 +++++++----------- .../components/hikvision/config_flow.py | 134 ++++++++ homeassistant/components/hikvision/const.py | 6 + .../components/hikvision/manifest.json | 2 + .../components/hikvision/strings.json | 36 ++ homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 4 +- requirements_test_all.txt | 3 + tests/components/hikvision/__init__.py | 14 + tests/components/hikvision/conftest.py | 94 ++++++ .../snapshots/test_binary_sensor.ambr | 101 ++++++ .../hikvision/test_binary_sensor.py | 300 +++++++++++++++++ .../components/hikvision/test_config_flow.py | 312 +++++++++++++++++ tests/components/hikvision/test_init.py | 62 ++++ 16 files changed, 1278 insertions(+), 194 deletions(-) create mode 100644 homeassistant/components/hikvision/config_flow.py create mode 100644 homeassistant/components/hikvision/const.py create mode 100644 homeassistant/components/hikvision/strings.json create mode 100644 tests/components/hikvision/__init__.py create mode 100644 tests/components/hikvision/conftest.py create mode 100644 tests/components/hikvision/snapshots/test_binary_sensor.ambr create mode 100644 tests/components/hikvision/test_binary_sensor.py create mode 100644 tests/components/hikvision/test_config_flow.py create mode 100644 tests/components/hikvision/test_init.py diff --git a/CODEOWNERS b/CODEOWNERS index 093fe2f0e68..629bed98fa4 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -665,6 +665,7 @@ build.json @home-assistant/supervisor /homeassistant/components/here_travel_time/ @eifinger /tests/components/here_travel_time/ @eifinger /homeassistant/components/hikvision/ @mezz64 +/tests/components/hikvision/ @mezz64 /homeassistant/components/hikvisioncam/ @fbradyirl /homeassistant/components/hisense_aehw4a1/ @bannhead /tests/components/hisense_aehw4a1/ @bannhead diff --git a/homeassistant/components/hikvision/__init__.py b/homeassistant/components/hikvision/__init__.py index dbf7991b3c4..1a7d042544e 100644 --- a/homeassistant/components/hikvision/__init__.py +++ b/homeassistant/components/hikvision/__init__.py @@ -1 +1,87 @@ -"""The hikvision component.""" +"""The Hikvision integration.""" + +from __future__ import annotations + +from dataclasses import dataclass +import logging + +from pyhik.hikvision import HikCamera +import requests + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + CONF_HOST, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, + Platform, +) +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryNotReady + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS = [Platform.BINARY_SENSOR] + + +@dataclass +class HikvisionData: + """Data class for Hikvision runtime data.""" + + camera: HikCamera + device_id: str + device_name: str + device_type: str + + +type HikvisionConfigEntry = ConfigEntry[HikvisionData] + + +async def async_setup_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool: + """Set up Hikvision from a config entry.""" + host = entry.data[CONF_HOST] + port = entry.data[CONF_PORT] + username = entry.data[CONF_USERNAME] + password = entry.data[CONF_PASSWORD] + ssl = entry.data[CONF_SSL] + + protocol = "https" if ssl else "http" + url = f"{protocol}://{host}" + + try: + camera = await hass.async_add_executor_job( + HikCamera, url, port, username, password + ) + except requests.exceptions.RequestException as err: + raise ConfigEntryNotReady(f"Unable to connect to {host}") from err + + device_id = camera.get_id() + if device_id is None: + raise ConfigEntryNotReady(f"Unable to get device ID from {host}") + + device_name = camera.get_name or host + device_type = camera.get_type or "Camera" + + entry.runtime_data = HikvisionData( + camera=camera, + device_id=device_id, + device_name=device_name, + device_type=device_type, + ) + + # Start the event stream + await hass.async_add_executor_job(camera.start_stream) + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: HikvisionConfigEntry) -> bool: + """Unload a config entry.""" + if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS): + # Stop the event stream + await hass.async_add_executor_job(entry.runtime_data.camera.disconnect) + + return unload_ok diff --git a/homeassistant/components/hikvision/binary_sensor.py b/homeassistant/components/hikvision/binary_sensor.py index 76cca5079e4..f0917c769bf 100644 --- a/homeassistant/components/hikvision/binary_sensor.py +++ b/homeassistant/components/hikvision/binary_sensor.py @@ -2,10 +2,9 @@ from __future__ import annotations -from datetime import timedelta import logging +from typing import Any -from pyhik.hikvision import HikCamera import voluptuous as vol from homeassistant.components.binary_sensor import ( @@ -13,6 +12,7 @@ from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, BinarySensorEntity, ) +from homeassistant.config_entries import SOURCE_IMPORT from homeassistant.const import ( ATTR_LAST_TRIP_TIME, CONF_CUSTOMIZE, @@ -23,27 +23,27 @@ from homeassistant.const import ( CONF_PORT, CONF_SSL, CONF_USERNAME, - EVENT_HOMEASSISTANT_START, - EVENT_HOMEASSISTANT_STOP, ) -from homeassistant.core import HomeAssistant -from homeassistant.helpers import config_validation as cv -from homeassistant.helpers.entity_platform import AddEntitiesCallback -from homeassistant.helpers.event import track_point_in_utc_time +from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.helpers import config_validation as cv, issue_registry as ir +from homeassistant.helpers.device_registry import DeviceInfo +from homeassistant.helpers.entity_platform import ( + AddConfigEntryEntitiesCallback, + AddEntitiesCallback, +) from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from homeassistant.util.dt import utcnow -_LOGGER = logging.getLogger(__name__) +from . import HikvisionConfigEntry +from .const import DEFAULT_PORT, DOMAIN CONF_IGNORED = "ignored" -DEFAULT_PORT = 80 -DEFAULT_IGNORED = False DEFAULT_DELAY = 0 +DEFAULT_IGNORED = False -ATTR_DELAY = "delay" - -DEVICE_CLASS_MAP = { +# Device class mapping for Hikvision event types +DEVICE_CLASS_MAP: dict[str, BinarySensorDeviceClass | None] = { "Motion": BinarySensorDeviceClass.MOTION, "Line Crossing": BinarySensorDeviceClass.MOTION, "Field Detection": BinarySensorDeviceClass.MOTION, @@ -67,6 +67,8 @@ DEVICE_CLASS_MAP = { "Entering Region": BinarySensorDeviceClass.MOTION, } +_LOGGER = logging.getLogger(__name__) + CUSTOMIZE_SCHEMA = vol.Schema( { vol.Optional(CONF_IGNORED, default=DEFAULT_IGNORED): cv.boolean, @@ -88,214 +90,144 @@ PLATFORM_SCHEMA = BINARY_SENSOR_PLATFORM_SCHEMA.extend( } ) +PARALLEL_UPDATES = 0 -def setup_platform( + +async def async_setup_platform( hass: HomeAssistant, config: ConfigType, - add_entities: AddEntitiesCallback, + async_add_entities: AddEntitiesCallback, discovery_info: DiscoveryInfoType | None = None, ) -> None: - """Set up the Hikvision binary sensor devices.""" - name = config.get(CONF_NAME) - host = config[CONF_HOST] - port = config[CONF_PORT] - username = config[CONF_USERNAME] - password = config[CONF_PASSWORD] + """Set up the Hikvision binary sensor platform from YAML.""" + # Trigger the import flow to migrate YAML config to config entry + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=config + ) - customize = config[CONF_CUSTOMIZE] - - protocol = "https" if config[CONF_SSL] else "http" - - url = f"{protocol}://{host}" - - data = HikvisionData(hass, url, port, name, username, password) - - if data.sensors is None: - _LOGGER.error("Hikvision event stream has no data, unable to set up") + if ( + result.get("type") is FlowResultType.ABORT + and result.get("reason") != "already_configured" + ): + ir.async_create_issue( + hass, + DOMAIN, + f"deprecated_yaml_import_issue_{result.get('reason')}", + is_fixable=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key="deprecated_yaml_import_issue", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": "Hikvision", + }, + ) return - entities = [] - - for sensor, channel_list in data.sensors.items(): - for channel in channel_list: - # Build sensor name, then parse customize config. - if data.type == "NVR": - sensor_name = f"{sensor.replace(' ', '_')}_{channel[1]}" - else: - sensor_name = sensor.replace(" ", "_") - - custom = customize.get(sensor_name.lower(), {}) - ignore = custom.get(CONF_IGNORED) - delay = custom.get(CONF_DELAY) - - _LOGGER.debug( - "Entity: %s - %s, Options - Ignore: %s, Delay: %s", - data.name, - sensor_name, - ignore, - delay, - ) - if not ignore: - entities.append( - HikvisionBinarySensor(hass, sensor, channel[1], data, delay) - ) - - add_entities(entities) + ir.async_create_issue( + hass, + HOMEASSISTANT_DOMAIN, + f"deprecated_yaml_{DOMAIN}", + is_fixable=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key="deprecated_yaml", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": "Hikvision", + }, + ) -class HikvisionData: - """Hikvision device event stream object.""" +async def async_setup_entry( + hass: HomeAssistant, + entry: HikvisionConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Hikvision binary sensors from a config entry.""" + data = entry.runtime_data + camera = data.camera - def __init__(self, hass, url, port, name, username, password): - """Initialize the data object.""" - self._url = url - self._port = port - self._name = name - self._username = username - self._password = password + sensors = camera.current_event_states + if sensors is None or not sensors: + _LOGGER.warning("Hikvision device has no sensors available") + return - # Establish camera - self.camdata = HikCamera(self._url, self._port, self._username, self._password) - - if self._name is None: - self._name = self.camdata.get_name - - hass.bus.listen_once(EVENT_HOMEASSISTANT_STOP, self.stop_hik) - hass.bus.listen_once(EVENT_HOMEASSISTANT_START, self.start_hik) - - def stop_hik(self, event): - """Shutdown Hikvision subscriptions and subscription thread on exit.""" - self.camdata.disconnect() - - def start_hik(self, event): - """Start Hikvision event stream thread.""" - self.camdata.start_stream() - - @property - def sensors(self): - """Return list of available sensors and their states.""" - return self.camdata.current_event_states - - @property - def cam_id(self): - """Return device id.""" - return self.camdata.get_id - - @property - def name(self): - """Return device name.""" - return self._name - - @property - def type(self): - """Return device type.""" - return self.camdata.get_type - - def get_attributes(self, sensor, channel): - """Return attribute list for sensor/channel.""" - return self.camdata.fetch_attributes(sensor, channel) + async_add_entities( + HikvisionBinarySensor( + entry=entry, + sensor_type=sensor_type, + channel=channel_info[1], + ) + for sensor_type, channel_list in sensors.items() + for channel_info in channel_list + ) class HikvisionBinarySensor(BinarySensorEntity): """Representation of a Hikvision binary sensor.""" + _attr_has_entity_name = True _attr_should_poll = False - def __init__(self, hass, sensor, channel, cam, delay): - """Initialize the binary_sensor.""" - self._hass = hass - self._cam = cam - self._sensor = sensor + def __init__( + self, + entry: HikvisionConfigEntry, + sensor_type: str, + channel: int, + ) -> None: + """Initialize the binary sensor.""" + self._data = entry.runtime_data + self._camera = self._data.camera + self._sensor_type = sensor_type self._channel = channel - if self._cam.type == "NVR": - self._name = f"{self._cam.name} {sensor} {channel}" + # Build unique ID + self._attr_unique_id = f"{self._data.device_id}_{sensor_type}_{channel}" + + # Build entity name based on device type + if self._data.device_type == "NVR": + self._attr_name = f"{sensor_type} {channel}" else: - self._name = f"{self._cam.name} {sensor}" + self._attr_name = sensor_type - self._id = f"{self._cam.cam_id}.{sensor}.{channel}" + # Device info for device registry + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, self._data.device_id)}, + name=self._data.device_name, + manufacturer="Hikvision", + model=self._data.device_type, + ) - if delay is None: - self._delay = 0 - else: - self._delay = delay + # Set device class + self._attr_device_class = DEVICE_CLASS_MAP.get(sensor_type) - self._timer = None + # Callback ID for pyhik + self._callback_id = f"{self._data.device_id}.{sensor_type}.{channel}" - # Register callback function with pyHik - self._cam.camdata.add_update_callback(self._update_callback, self._id) - - def _sensor_state(self): - """Extract sensor state.""" - return self._cam.get_attributes(self._sensor, self._channel)[0] - - def _sensor_last_update(self): - """Extract sensor last update time.""" - return self._cam.get_attributes(self._sensor, self._channel)[3] + def _get_sensor_attributes(self) -> tuple[bool, Any, Any, Any]: + """Get sensor attributes from camera.""" + return self._camera.fetch_attributes(self._sensor_type, self._channel) @property - def name(self): - """Return the name of the Hikvision sensor.""" - return self._name - - @property - def unique_id(self): - """Return a unique ID.""" - return self._id - - @property - def is_on(self): + def is_on(self) -> bool: """Return true if sensor is on.""" - return self._sensor_state() + return self._get_sensor_attributes()[0] @property - def device_class(self): - """Return the class of this sensor, from DEVICE_CLASSES.""" - try: - return DEVICE_CLASS_MAP[self._sensor] - except KeyError: - # Sensor must be unknown to us, add as generic - return None - - @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the state attributes.""" - attr = {ATTR_LAST_TRIP_TIME: self._sensor_last_update()} + attrs = self._get_sensor_attributes() + return {ATTR_LAST_TRIP_TIME: attrs[3]} - if self._delay != 0: - attr[ATTR_DELAY] = self._delay + async def async_added_to_hass(self) -> None: + """Register callback when entity is added.""" + await super().async_added_to_hass() - return attr + # Register callback with pyhik + self._camera.add_update_callback(self._update_callback, self._callback_id) - def _update_callback(self, msg): - """Update the sensor's state, if needed.""" - _LOGGER.debug("Callback signal from: %s", msg) - - if self._delay > 0 and not self.is_on: - # Set timer to wait until updating the state - def _delay_update(now): - """Timer callback for sensor update.""" - _LOGGER.debug( - "%s Called delayed (%ssec) update", self._name, self._delay - ) - self.schedule_update_ha_state() - self._timer = None - - if self._timer is not None: - self._timer() - self._timer = None - - self._timer = track_point_in_utc_time( - self._hass, _delay_update, utcnow() + timedelta(seconds=self._delay) - ) - - elif self._delay > 0 and self.is_on: - # For delayed sensors kill any callbacks on true events and update - if self._timer is not None: - self._timer() - self._timer = None - - self.schedule_update_ha_state() - - else: - self.schedule_update_ha_state() + @callback + def _update_callback(self, msg: str) -> None: + """Update the sensor's state when callback is triggered.""" + self.async_write_ha_state() diff --git a/homeassistant/components/hikvision/config_flow.py b/homeassistant/components/hikvision/config_flow.py new file mode 100644 index 00000000000..f5f89c1646a --- /dev/null +++ b/homeassistant/components/hikvision/config_flow.py @@ -0,0 +1,134 @@ +"""Config flow for Hikvision integration.""" + +from __future__ import annotations + +import logging +from typing import Any + +from pyhik.hikvision import HikCamera +import requests +import voluptuous as vol + +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import ( + CONF_HOST, + CONF_NAME, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, +) +from homeassistant.helpers.typing import ConfigType + +from .const import DEFAULT_PORT, DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +class HikvisionConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Hikvision.""" + + VERSION = 1 + MINOR_VERSION = 1 + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the initial step.""" + errors: dict[str, str] = {} + + if user_input is not None: + host = user_input[CONF_HOST] + port = user_input[CONF_PORT] + username = user_input[CONF_USERNAME] + password = user_input[CONF_PASSWORD] + ssl = user_input[CONF_SSL] + + protocol = "https" if ssl else "http" + url = f"{protocol}://{host}" + + try: + camera = await self.hass.async_add_executor_job( + HikCamera, url, port, username, password + ) + device_id = camera.get_id() + device_name = camera.get_name + except requests.exceptions.RequestException: + _LOGGER.exception("Error connecting to Hikvision device") + errors["base"] = "cannot_connect" + else: + if device_id is None: + errors["base"] = "cannot_connect" + else: + await self.async_set_unique_id(device_id) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=device_name or host, + data={ + CONF_HOST: host, + CONF_PORT: port, + CONF_USERNAME: username, + CONF_PASSWORD: password, + CONF_SSL: ssl, + }, + ) + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema( + { + vol.Required(CONF_HOST): str, + vol.Required(CONF_PORT, default=DEFAULT_PORT): int, + vol.Required(CONF_USERNAME): str, + vol.Required(CONF_PASSWORD): str, + vol.Required(CONF_SSL, default=False): bool, + } + ), + errors=errors, + ) + + async def async_step_import(self, import_data: ConfigType) -> ConfigFlowResult: + """Handle import from configuration.yaml.""" + host = import_data[CONF_HOST] + port = import_data.get(CONF_PORT, DEFAULT_PORT) + username = import_data[CONF_USERNAME] + password = import_data[CONF_PASSWORD] + ssl = import_data.get(CONF_SSL, False) + name = import_data.get(CONF_NAME) + + protocol = "https" if ssl else "http" + url = f"{protocol}://{host}" + + try: + camera = await self.hass.async_add_executor_job( + HikCamera, url, port, username, password + ) + device_id = camera.get_id() + device_name = camera.get_name + except requests.exceptions.RequestException: + _LOGGER.exception( + "Error connecting to Hikvision device during import, aborting" + ) + return self.async_abort(reason="cannot_connect") + + if device_id is None: + return self.async_abort(reason="cannot_connect") + + await self.async_set_unique_id(device_id) + self._abort_if_unique_id_configured() + + _LOGGER.warning( + "Importing Hikvision config from configuration.yaml for %s", host + ) + + return self.async_create_entry( + title=name or device_name or host, + data={ + CONF_HOST: host, + CONF_PORT: port, + CONF_USERNAME: username, + CONF_PASSWORD: password, + CONF_SSL: ssl, + }, + ) diff --git a/homeassistant/components/hikvision/const.py b/homeassistant/components/hikvision/const.py new file mode 100644 index 00000000000..14e6a1808b7 --- /dev/null +++ b/homeassistant/components/hikvision/const.py @@ -0,0 +1,6 @@ +"""Constants for the Hikvision integration.""" + +DOMAIN = "hikvision" + +# Default values +DEFAULT_PORT = 80 diff --git a/homeassistant/components/hikvision/manifest.json b/homeassistant/components/hikvision/manifest.json index a0832732105..84c9240192e 100644 --- a/homeassistant/components/hikvision/manifest.json +++ b/homeassistant/components/hikvision/manifest.json @@ -2,7 +2,9 @@ "domain": "hikvision", "name": "Hikvision", "codeowners": ["@mezz64"], + "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/hikvision", + "integration_type": "device", "iot_class": "local_push", "loggers": ["pyhik"], "quality_scale": "legacy", diff --git a/homeassistant/components/hikvision/strings.json b/homeassistant/components/hikvision/strings.json new file mode 100644 index 00000000000..4501113d309 --- /dev/null +++ b/homeassistant/components/hikvision/strings.json @@ -0,0 +1,36 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]" + }, + "step": { + "user": { + "data": { + "host": "[%key:common::config_flow::data::host%]", + "password": "[%key:common::config_flow::data::password%]", + "port": "[%key:common::config_flow::data::port%]", + "ssl": "Use SSL", + "username": "[%key:common::config_flow::data::username%]" + }, + "data_description": { + "host": "The hostname or IP address of your Hikvision device", + "password": "The password for your Hikvision device", + "port": "The port number for the device (default is 80)", + "ssl": "Enable if your device uses HTTPS", + "username": "The username for your Hikvision device" + }, + "description": "Enter your Hikvision device connection details.", + "title": "Set up Hikvision device" + } + } + }, + "issues": { + "deprecated_yaml_import_issue": { + "description": "Configuring {integration_title} using YAML is deprecated and the import failed. Please remove the `{domain}` entry from your `configuration.yaml` file and set up the integration manually.", + "title": "YAML import failed" + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 23351374cf4..8b78ae3ac44 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -278,6 +278,7 @@ FLOWS = { "harmony", "heos", "here_travel_time", + "hikvision", "hisense_aehw4a1", "hive", "hko", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index ca0b78a86b9..1e75aed91dc 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -2714,8 +2714,8 @@ "name": "Hikvision", "integrations": { "hikvision": { - "integration_type": "hub", - "config_flow": false, + "integration_type": "device", + "config_flow": true, "iot_class": "local_push", "name": "Hikvision" }, diff --git a/requirements_test_all.txt b/requirements_test_all.txt index 4ba60a5ada7..5fbdbb36bf8 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1582,6 +1582,9 @@ pyDuotecno==2024.10.1 # homeassistant.components.electrasmart pyElectra==1.2.4 +# homeassistant.components.hikvision +pyHik==0.3.2 + # homeassistant.components.homee pyHomee==1.3.8 diff --git a/tests/components/hikvision/__init__.py b/tests/components/hikvision/__init__.py new file mode 100644 index 00000000000..15c5f6c5745 --- /dev/null +++ b/tests/components/hikvision/__init__.py @@ -0,0 +1,14 @@ +"""Common test tools for the Hikvision integration.""" + +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry + + +async def setup_integration( + hass: HomeAssistant, mock_config_entry: MockConfigEntry +) -> None: + """Set up the Hikvision integration for testing.""" + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() diff --git a/tests/components/hikvision/conftest.py b/tests/components/hikvision/conftest.py new file mode 100644 index 00000000000..13b000c46ca --- /dev/null +++ b/tests/components/hikvision/conftest.py @@ -0,0 +1,94 @@ +"""Common fixtures for the Hikvision tests.""" + +from collections.abc import Generator +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from homeassistant.components.hikvision.const import DOMAIN +from homeassistant.const import ( + CONF_HOST, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, +) +from homeassistant.core import HomeAssistant + +from . import setup_integration + +from tests.common import MockConfigEntry + +TEST_HOST = "192.168.1.100" +TEST_PORT = 80 +TEST_USERNAME = "admin" +TEST_PASSWORD = "password123" +TEST_DEVICE_ID = "DS-2CD2142FWD-I20170101AAAA" +TEST_DEVICE_NAME = "Front Camera" + + +@pytest.fixture +def mock_setup_entry() -> Generator[AsyncMock]: + """Override async_setup_entry.""" + with patch( + "homeassistant.components.hikvision.async_setup_entry", return_value=True + ) as mock_setup_entry: + yield mock_setup_entry + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Return the default mocked config entry.""" + return MockConfigEntry( + title=TEST_DEVICE_NAME, + domain=DOMAIN, + version=1, + minor_version=1, + data={ + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + unique_id=TEST_DEVICE_ID, + ) + + +@pytest.fixture +def mock_hikcamera() -> Generator[MagicMock]: + """Return a mocked HikCamera.""" + with ( + patch( + "homeassistant.components.hikvision.HikCamera", + autospec=True, + ) as hikcamera_mock, + patch( + "homeassistant.components.hikvision.config_flow.HikCamera", + new=hikcamera_mock, + ), + ): + camera = hikcamera_mock.return_value + camera.get_id.return_value = TEST_DEVICE_ID + camera.get_name = TEST_DEVICE_NAME + camera.get_type = "Camera" + camera.current_event_states = { + "Motion": [(True, 1)], + "Line Crossing": [(False, 1)], + } + camera.fetch_attributes.return_value = ( + False, + None, + None, + "2024-01-01T00:00:00Z", + ) + yield hikcamera_mock + + +@pytest.fixture +async def init_integration( + hass: HomeAssistant, mock_config_entry: MockConfigEntry, mock_hikcamera: MagicMock +) -> MockConfigEntry: + """Set up the Hikvision integration for testing.""" + await setup_integration(hass, mock_config_entry) + return mock_config_entry diff --git a/tests/components/hikvision/snapshots/test_binary_sensor.ambr b/tests/components/hikvision/snapshots/test_binary_sensor.ambr new file mode 100644 index 00000000000..2d77654cd58 --- /dev/null +++ b/tests/components/hikvision/snapshots/test_binary_sensor.ambr @@ -0,0 +1,101 @@ +# serializer version: 1 +# name: test_all_entities[binary_sensor.front_camera_line_crossing-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'binary_sensor', + 'entity_category': None, + 'entity_id': 'binary_sensor.front_camera_line_crossing', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Line Crossing', + 'platform': 'hikvision', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': 'DS-2CD2142FWD-I20170101AAAA_Line Crossing_1', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[binary_sensor.front_camera_line_crossing-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'device_class': 'motion', + 'friendly_name': 'Front Camera Line Crossing', + 'last_tripped_time': '2024-01-01T00:00:00Z', + }), + 'context': , + 'entity_id': 'binary_sensor.front_camera_line_crossing', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'off', + }) +# --- +# name: test_all_entities[binary_sensor.front_camera_motion-entry] + EntityRegistryEntrySnapshot({ + 'aliases': set({ + }), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'binary_sensor', + 'entity_category': None, + 'entity_id': 'binary_sensor.front_camera_motion', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Motion', + 'platform': 'hikvision', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': 'DS-2CD2142FWD-I20170101AAAA_Motion_1', + 'unit_of_measurement': None, + }) +# --- +# name: test_all_entities[binary_sensor.front_camera_motion-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + 'device_class': 'motion', + 'friendly_name': 'Front Camera Motion', + 'last_tripped_time': '2024-01-01T00:00:00Z', + }), + 'context': , + 'entity_id': 'binary_sensor.front_camera_motion', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'off', + }) +# --- diff --git a/tests/components/hikvision/test_binary_sensor.py b/tests/components/hikvision/test_binary_sensor.py new file mode 100644 index 00000000000..c89e77652db --- /dev/null +++ b/tests/components/hikvision/test_binary_sensor.py @@ -0,0 +1,300 @@ +"""Test Hikvision binary sensors.""" + +from unittest.mock import MagicMock + +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant.components.binary_sensor import BinarySensorDeviceClass +from homeassistant.components.hikvision.const import DOMAIN +from homeassistant.const import ( + ATTR_DEVICE_CLASS, + ATTR_LAST_TRIP_TIME, + CONF_HOST, + CONF_NAME, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, + STATE_OFF, +) +from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant +from homeassistant.helpers import ( + device_registry as dr, + entity_registry as er, + issue_registry as ir, +) +from homeassistant.setup import async_setup_component + +from . import setup_integration +from .conftest import ( + TEST_DEVICE_ID, + TEST_DEVICE_NAME, + TEST_HOST, + TEST_PASSWORD, + TEST_PORT, + TEST_USERNAME, +) + +from tests.common import MockConfigEntry, snapshot_platform + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_all_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + entity_registry: er.EntityRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test all binary sensor entities.""" + await setup_integration(hass, mock_config_entry) + await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id) + + +async def test_binary_sensors_created( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test binary sensors are created for each event type.""" + await setup_integration(hass, mock_config_entry) + + # Check Motion sensor (camera type doesn't include channel in name) + state = hass.states.get("binary_sensor.front_camera_motion") + assert state is not None + assert state.state == STATE_OFF + assert state.attributes.get(ATTR_DEVICE_CLASS) == BinarySensorDeviceClass.MOTION + assert ATTR_LAST_TRIP_TIME in state.attributes + + # Check Line Crossing sensor + state = hass.states.get("binary_sensor.front_camera_line_crossing") + assert state is not None + assert state.state == STATE_OFF + assert state.attributes.get(ATTR_DEVICE_CLASS) == BinarySensorDeviceClass.MOTION + + +async def test_binary_sensor_device_info( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, + device_registry: dr.DeviceRegistry, +) -> None: + """Test binary sensors are linked to device.""" + await setup_integration(hass, mock_config_entry) + + device_entry = device_registry.async_get_device( + identifiers={(DOMAIN, TEST_DEVICE_ID)} + ) + assert device_entry is not None + assert device_entry.name == TEST_DEVICE_NAME + assert device_entry.manufacturer == "Hikvision" + assert device_entry.model == "Camera" + + +async def test_binary_sensor_callback_registered( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test that callback is registered with pyhik.""" + await setup_integration(hass, mock_config_entry) + + # Verify callback was registered for each sensor + assert mock_hikcamera.return_value.add_update_callback.call_count == 2 + + +async def test_binary_sensor_no_sensors( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test setup when device has no sensors.""" + mock_hikcamera.return_value.current_event_states = None + + await setup_integration(hass, mock_config_entry) + + # No binary sensors should be created + states = hass.states.async_entity_ids("binary_sensor") + assert len(states) == 0 + + +async def test_binary_sensor_nvr_device( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test binary sensor naming for NVR devices.""" + mock_hikcamera.return_value.get_type = "NVR" + mock_hikcamera.return_value.current_event_states = { + "Motion": [(True, 1), (False, 2)], + } + + await setup_integration(hass, mock_config_entry) + + # NVR sensors should include channel number in name + state = hass.states.get("binary_sensor.front_camera_motion_1") + assert state is not None + + state = hass.states.get("binary_sensor.front_camera_motion_2") + assert state is not None + + +async def test_binary_sensor_state_on( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test binary sensor state when on.""" + mock_hikcamera.return_value.fetch_attributes.return_value = ( + True, + None, + None, + "2024-01-01T12:00:00Z", + ) + + await setup_integration(hass, mock_config_entry) + + state = hass.states.get("binary_sensor.front_camera_motion") + assert state is not None + assert state.state == "on" + + +async def test_binary_sensor_device_class_unknown( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test binary sensor with unknown device class.""" + mock_hikcamera.return_value.current_event_states = { + "Unknown Event": [(False, 1)], + } + + await setup_integration(hass, mock_config_entry) + + state = hass.states.get("binary_sensor.front_camera_unknown_event") + assert state is not None + assert state.attributes.get(ATTR_DEVICE_CLASS) is None + + +async def test_yaml_import_creates_deprecation_issue( + hass: HomeAssistant, + mock_hikcamera: MagicMock, + issue_registry: ir.IssueRegistry, +) -> None: + """Test YAML import creates deprecation issue.""" + await async_setup_component( + hass, + "binary_sensor", + { + "binary_sensor": { + "platform": DOMAIN, + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + } + }, + ) + await hass.async_block_till_done() + + # Check that deprecation issue was created in homeassistant domain + issue = issue_registry.async_get_issue( + HOMEASSISTANT_DOMAIN, f"deprecated_yaml_{DOMAIN}" + ) + assert issue is not None + assert issue.severity == ir.IssueSeverity.WARNING + + +async def test_yaml_import_with_name( + hass: HomeAssistant, + mock_hikcamera: MagicMock, +) -> None: + """Test YAML import uses custom name for config entry.""" + await async_setup_component( + hass, + "binary_sensor", + { + "binary_sensor": { + "platform": DOMAIN, + CONF_NAME: "Custom Camera Name", + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + } + }, + ) + await hass.async_block_till_done() + + # Check that the config entry was created with the custom name + entries = hass.config_entries.async_entries(DOMAIN) + assert len(entries) == 1 + assert entries[0].title == "Custom Camera Name" + + +async def test_yaml_import_abort_creates_issue( + hass: HomeAssistant, + mock_hikcamera: MagicMock, + issue_registry: ir.IssueRegistry, +) -> None: + """Test YAML import creates issue when import is aborted.""" + mock_hikcamera.return_value.get_id.return_value = None + + await async_setup_component( + hass, + "binary_sensor", + { + "binary_sensor": { + "platform": DOMAIN, + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + } + }, + ) + await hass.async_block_till_done() + + # Check that import failure issue was created + issue = issue_registry.async_get_issue( + DOMAIN, "deprecated_yaml_import_issue_cannot_connect" + ) + assert issue is not None + assert issue.severity == ir.IssueSeverity.WARNING + + +async def test_binary_sensor_update_callback( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test binary sensor state updates via callback.""" + await setup_integration(hass, mock_config_entry) + + state = hass.states.get("binary_sensor.front_camera_motion") + assert state is not None + assert state.state == STATE_OFF + + # Simulate state change via callback + mock_hikcamera.return_value.fetch_attributes.return_value = ( + True, + None, + None, + "2024-01-01T12:00:00Z", + ) + + # Get the registered callback and call it + add_callback_call = mock_hikcamera.return_value.add_update_callback.call_args_list[ + 0 + ] + callback_func = add_callback_call[0][0] + callback_func("motion detected") + + # Verify state was updated + state = hass.states.get("binary_sensor.front_camera_motion") + assert state is not None + assert state.state == "on" diff --git a/tests/components/hikvision/test_config_flow.py b/tests/components/hikvision/test_config_flow.py new file mode 100644 index 00000000000..5f1766a8d99 --- /dev/null +++ b/tests/components/hikvision/test_config_flow.py @@ -0,0 +1,312 @@ +"""Test the Hikvision config flow.""" + +from unittest.mock import AsyncMock, MagicMock + +import requests + +from homeassistant.components.hikvision.const import DOMAIN +from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER +from homeassistant.const import ( + CONF_HOST, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, +) +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from .conftest import ( + TEST_DEVICE_ID, + TEST_DEVICE_NAME, + TEST_HOST, + TEST_PASSWORD, + TEST_PORT, + TEST_USERNAME, +) + +from tests.common import MockConfigEntry + + +async def test_form( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test we get the form and can create entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert result["errors"] == {} + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == TEST_DEVICE_NAME + assert result["result"].unique_id == TEST_DEVICE_ID + assert result["data"] == { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + } + + +async def test_form_cannot_connect( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test we handle cannot connect error and can recover.""" + mock_hikcamera.return_value.get_id.return_value = None + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + + # Recover from error + mock_hikcamera.return_value.get_id.return_value = TEST_DEVICE_ID + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["result"].unique_id == TEST_DEVICE_ID + + +async def test_form_exception( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test we handle exception during connection and can recover.""" + mock_hikcamera.side_effect = requests.exceptions.RequestException( + "Connection failed" + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + + # Recover from error + mock_hikcamera.side_effect = None + mock_hikcamera.return_value.get_id.return_value = TEST_DEVICE_ID + mock_hikcamera.return_value.get_name = TEST_DEVICE_NAME + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["result"].unique_id == TEST_DEVICE_ID + + +async def test_form_already_configured( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test we handle already configured devices.""" + mock_config_entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_import_flow( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test YAML import flow creates config entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == TEST_DEVICE_NAME + assert result["result"].unique_id == TEST_DEVICE_ID + assert result["data"] == { + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + } + + +async def test_import_flow_with_defaults( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test YAML import flow uses default values.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: TEST_HOST, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["result"].unique_id == TEST_DEVICE_ID + # Default port (80) and SSL (False) should be used + assert result["data"][CONF_PORT] == 80 + assert result["data"][CONF_SSL] is False + + +async def test_import_flow_cannot_connect( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test YAML import flow aborts on connection error.""" + mock_hikcamera.side_effect = requests.exceptions.RequestException( + "Connection failed" + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_import_flow_no_device_id( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, +) -> None: + """Test YAML import flow aborts when device_id is None.""" + mock_hikcamera.return_value.get_id.return_value = None + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + + +async def test_import_flow_already_configured( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_hikcamera: MagicMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test YAML import flow aborts when device is already configured.""" + mock_config_entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={ + CONF_HOST: TEST_HOST, + CONF_PORT: TEST_PORT, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, + CONF_SSL: False, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" diff --git a/tests/components/hikvision/test_init.py b/tests/components/hikvision/test_init.py new file mode 100644 index 00000000000..3bf3136c648 --- /dev/null +++ b/tests/components/hikvision/test_init.py @@ -0,0 +1,62 @@ +"""Test Hikvision integration setup and unload.""" + +from unittest.mock import MagicMock + +import requests + +from homeassistant.config_entries import ConfigEntryState +from homeassistant.core import HomeAssistant + +from . import setup_integration + +from tests.common import MockConfigEntry + + +async def test_setup_and_unload_entry( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test successful setup and unload of config entry.""" + await setup_integration(hass, mock_config_entry) + + assert mock_config_entry.state is ConfigEntryState.LOADED + mock_hikcamera.return_value.start_stream.assert_called_once() + + await hass.config_entries.async_unload(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.NOT_LOADED + mock_hikcamera.return_value.disconnect.assert_called_once() + + +async def test_setup_entry_connection_error( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test setup fails on connection error.""" + mock_hikcamera.side_effect = requests.exceptions.RequestException( + "Connection failed" + ) + + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY + + +async def test_setup_entry_no_device_id( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_hikcamera: MagicMock, +) -> None: + """Test setup fails when device_id is None.""" + mock_hikcamera.return_value.get_id.return_value = None + + mock_config_entry.add_to_hass(hass) + await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY