Migrate Hikvision integration to config flow (#158279)

Co-authored-by: Kamil Breguła <mik-laj@users.noreply.github.com>

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

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

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()

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,
+            },
+        )

homeassistant/components/hikvision/const.py

@@ -0,0 +1,6 @@
+"""Constants for the Hikvision integration."""
+
+DOMAIN = "hikvision"
+
+# Default values
+DEFAULT_PORT = 80

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",

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"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -278,6 +278,7 @@ FLOWS = {
         "harmony",
         "heos",
         "here_travel_time",
+        "hikvision",
         "hisense_aehw4a1",
         "hive",
         "hko",

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"
         },

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
 

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()

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

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': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    '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': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <BinarySensorDeviceClass.MOTION: 'motion'>,
+    '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': <ANY>,
+    'entity_id': 'binary_sensor.front_camera_line_crossing',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---
+# name: test_all_entities[binary_sensor.front_camera_motion-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': set({
+    }),
+    'area_id': None,
+    'capabilities': None,
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    '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': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'options': dict({
+    }),
+    'original_device_class': <BinarySensorDeviceClass.MOTION: 'motion'>,
+    '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': <ANY>,
+    'entity_id': 'binary_sensor.front_camera_motion',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'off',
+  })
+# ---

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"

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"

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