diff --git a/CODEOWNERS b/CODEOWNERS index c29f37bab5913..69fe3899c4ebd 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -1763,6 +1763,7 @@ build.json @home-assistant/supervisor /homeassistant/components/vilfo/ @ManneW /tests/components/vilfo/ @ManneW /homeassistant/components/vivotek/ @HarlemSquirrel +/tests/components/vivotek/ @HarlemSquirrel /homeassistant/components/vizio/ @raman325 /tests/components/vizio/ @raman325 /homeassistant/components/vlc_telnet/ @rodripf @MartinHjelmare diff --git a/homeassistant/components/vivotek/__init__.py b/homeassistant/components/vivotek/__init__.py index b5220b12a9b68..e8656096a9879 100644 --- a/homeassistant/components/vivotek/__init__.py +++ b/homeassistant/components/vivotek/__init__.py @@ -1 +1,70 @@ """The Vivotek camera component.""" + +import logging +from typing import Any + +from libpyvivotek.vivotek import VivotekCamera, VivotekCameraError + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import ( + CONF_AUTHENTICATION, + CONF_IP_ADDRESS, + CONF_PASSWORD, + CONF_PORT, + CONF_USERNAME, + CONF_VERIFY_SSL, + HTTP_DIGEST_AUTHENTICATION, + Platform, +) +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryError + +from .const import CONF_SECURITY_LEVEL + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS = [Platform.CAMERA] + +type VivotekConfigEntry = ConfigEntry[VivotekCamera] + + +def build_cam_client(data: dict[str, Any]) -> VivotekCamera: + """Build the Vivotek camera client from the provided configuration data.""" + return VivotekCamera( + host=data[CONF_IP_ADDRESS], + port=data[CONF_PORT], + verify_ssl=data[CONF_VERIFY_SSL], + usr=data[CONF_USERNAME], + pwd=data[CONF_PASSWORD], + digest_auth=(data[CONF_AUTHENTICATION] == HTTP_DIGEST_AUTHENTICATION), + sec_lvl=data[CONF_SECURITY_LEVEL], + ) + + +async def async_build_and_test_cam_client( + hass: HomeAssistant, data: dict[str, Any] +) -> VivotekCamera: + """Build the client and test if the provided configuration is valid.""" + cam_client = build_cam_client(data) + await hass.async_add_executor_job(cam_client.get_mac) + + return cam_client + + +async def async_setup_entry(hass: HomeAssistant, entry: VivotekConfigEntry) -> bool: + """Set up the Vivotek component from a config entry.""" + + try: + cam_client = await async_build_and_test_cam_client(hass, dict(entry.data)) + except VivotekCameraError as err: + raise ConfigEntryError("Failed to connect to camera") from err + + entry.runtime_data = cam_client + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: VivotekConfigEntry) -> bool: + """Unload a config entry.""" + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) diff --git a/homeassistant/components/vivotek/camera.py b/homeassistant/components/vivotek/camera.py index c044e99a82e24..5b22ba41349d1 100644 --- a/homeassistant/components/vivotek/camera.py +++ b/homeassistant/components/vivotek/camera.py @@ -2,7 +2,10 @@ from __future__ import annotations -from libpyvivotek import VivotekCamera +import logging +from typing import TYPE_CHECKING + +from libpyvivotek.vivotek import VivotekCamera import voluptuous as vol from homeassistant.components.camera import ( @@ -10,6 +13,7 @@ Camera, CameraEntityFeature, ) +from homeassistant.config_entries import SOURCE_IMPORT from homeassistant.const import ( CONF_AUTHENTICATION, CONF_IP_ADDRESS, @@ -21,18 +25,30 @@ HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION, ) -from homeassistant.core import HomeAssistant -from homeassistant.helpers import config_validation as cv -from homeassistant.helpers.entity_platform import AddEntitiesCallback +from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant +from homeassistant.data_entry_flow import FlowResultType +from homeassistant.helpers import config_validation as cv, issue_registry as ir +from homeassistant.helpers.entity_platform import ( + AddConfigEntryEntitiesCallback, + AddEntitiesCallback, +) from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -CONF_FRAMERATE = "framerate" -CONF_SECURITY_LEVEL = "security_level" -CONF_STREAM_PATH = "stream_path" +from . import VivotekConfigEntry +from .const import ( + CONF_FRAMERATE, + CONF_SECURITY_LEVEL, + CONF_STREAM_PATH, + DOMAIN, + INTEGRATION_TITLE, +) + +_LOGGER = logging.getLogger(__name__) DEFAULT_CAMERA_BRAND = "VIVOTEK" DEFAULT_NAME = "VIVOTEK Camera" DEFAULT_EVENT_0_KEY = "event_i0_enable" +DEFAULT_FRAMERATE = 2 DEFAULT_SECURITY_LEVEL = "admin" DEFAULT_STREAM_SOURCE = "live.sdp" @@ -47,34 +63,86 @@ ), vol.Optional(CONF_SSL, default=False): cv.boolean, vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean, - vol.Optional(CONF_FRAMERATE, default=2): cv.positive_int, + vol.Optional(CONF_FRAMERATE, default=DEFAULT_FRAMERATE): cv.positive_int, vol.Optional(CONF_SECURITY_LEVEL, default=DEFAULT_SECURITY_LEVEL): cv.string, vol.Optional(CONF_STREAM_PATH, default=DEFAULT_STREAM_SOURCE): cv.string, } ) -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 a Vivotek IP Camera.""" - creds = f"{config[CONF_USERNAME]}:{config[CONF_PASSWORD]}" - cam = VivotekCamera( - host=config[CONF_IP_ADDRESS], - port=(443 if config[CONF_SSL] else 80), - verify_ssl=config[CONF_VERIFY_SSL], - usr=config[CONF_USERNAME], - pwd=config[CONF_PASSWORD], - digest_auth=config[CONF_AUTHENTICATION] == HTTP_DIGEST_AUTHENTICATION, - sec_lvl=config[CONF_SECURITY_LEVEL], + """Set up the Vivotek camera platform.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data=config, + ) + 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')}", + breaks_in_ha_version="2026.6.0", + is_fixable=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key=f"deprecated_yaml_import_issue_{result.get('reason')}", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": INTEGRATION_TITLE, + }, + ) + return + + ir.async_create_issue( + hass, + HOMEASSISTANT_DOMAIN, + "deprecated_yaml", + breaks_in_ha_version="2026.6.0", + is_fixable=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key="deprecated_yaml", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": INTEGRATION_TITLE, + }, ) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: VivotekConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up the component from a config entry.""" + config = entry.data + creds = f"{config[CONF_USERNAME]}:{config[CONF_PASSWORD]}" stream_source = ( f"rtsp://{creds}@{config[CONF_IP_ADDRESS]}:554/{config[CONF_STREAM_PATH]}" ) - add_entities([VivotekCam(config, cam, stream_source)], True) + cam_client = entry.runtime_data + if TYPE_CHECKING: + assert entry.unique_id is not None + async_add_entities( + [ + VivotekCam( + cam_client, + stream_source, + entry.unique_id, + entry.options[CONF_FRAMERATE], + entry.title, + ) + ] + ) class VivotekCam(Camera): @@ -84,14 +152,19 @@ class VivotekCam(Camera): _attr_supported_features = CameraEntityFeature.STREAM def __init__( - self, config: ConfigType, cam: VivotekCamera, stream_source: str + self, + cam_client: VivotekCamera, + stream_source: str, + unique_id: str, + framerate: int, + name: str, ) -> None: """Initialize a Vivotek camera.""" super().__init__() - - self._cam = cam - self._attr_frame_interval = 1 / config[CONF_FRAMERATE] - self._attr_name = config[CONF_NAME] + self._cam = cam_client + self._attr_frame_interval = 1 / framerate + self._attr_unique_id = unique_id + self._attr_name = name self._stream_source = stream_source def camera_image( @@ -117,3 +190,4 @@ def enable_motion_detection(self) -> None: def update(self) -> None: """Update entity status.""" self._attr_model = self._cam.model_name + self._attr_available = self._attr_model is not None diff --git a/homeassistant/components/vivotek/config_flow.py b/homeassistant/components/vivotek/config_flow.py new file mode 100644 index 0000000000000..7d54d22e16043 --- /dev/null +++ b/homeassistant/components/vivotek/config_flow.py @@ -0,0 +1,181 @@ +"""Config flow for Vivotek IP cameras integration.""" + +import logging +from typing import Any + +from libpyvivotek.vivotek import SECURITY_LEVELS, VivotekCameraError +import voluptuous as vol + +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult, OptionsFlow +from homeassistant.const import ( + CONF_AUTHENTICATION, + CONF_IP_ADDRESS, + CONF_NAME, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, + CONF_VERIFY_SSL, + HTTP_BASIC_AUTHENTICATION, + HTTP_DIGEST_AUTHENTICATION, + UnitOfFrequency, +) +from homeassistant.core import callback +from homeassistant.helpers import config_validation as cv +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.selector import ( + NumberSelector, + NumberSelectorConfig, + SelectSelector, + SelectSelectorConfig, + SelectSelectorMode, +) + +from . import VivotekConfigEntry, build_cam_client +from .camera import DEFAULT_FRAMERATE, DEFAULT_NAME, DEFAULT_STREAM_SOURCE +from .const import CONF_FRAMERATE, CONF_SECURITY_LEVEL, CONF_STREAM_PATH, DOMAIN + +_LOGGER = logging.getLogger(__name__) + +DESCRIPTION_PLACEHOLDERS = { + "doc_url": "https://www.home-assistant.io/integrations/vivotek/" +} + +CONF_SCHEMA = vol.Schema( + { + vol.Required(CONF_IP_ADDRESS): cv.string, + vol.Required(CONF_PORT, default=80): cv.port, + vol.Required(CONF_USERNAME): cv.string, + vol.Required(CONF_PASSWORD): cv.string, + vol.Required(CONF_AUTHENTICATION, default=HTTP_BASIC_AUTHENTICATION): vol.In( + [HTTP_BASIC_AUTHENTICATION, HTTP_DIGEST_AUTHENTICATION] + ), + vol.Required(CONF_SSL, default=False): cv.boolean, + vol.Required(CONF_VERIFY_SSL, default=True): cv.boolean, + vol.Required(CONF_SECURITY_LEVEL): SelectSelector( + SelectSelectorConfig( + options=list(SECURITY_LEVELS.keys()), + mode=SelectSelectorMode.DROPDOWN, + translation_key="security_level", + sort=True, + ), + ), + vol.Required( + CONF_STREAM_PATH, + default=DEFAULT_STREAM_SOURCE, + ): cv.string, + } +) + +OPTIONS_SCHEMA = vol.Schema( + { + vol.Required(CONF_FRAMERATE, default=DEFAULT_FRAMERATE): NumberSelector( + NumberSelectorConfig(min=0, unit_of_measurement=UnitOfFrequency.HERTZ) + ), + } +) + + +class OptionsFlowHandler(OptionsFlow): + """Options flow.""" + + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Manage the options.""" + if user_input is not None: + return self.async_create_entry( + data=user_input, + ) + + return self.async_show_form( + step_id="init", + data_schema=self.add_suggested_values_to_schema( + OPTIONS_SCHEMA, self.config_entry.options + ), + ) + + +class VivotekConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Vivotek IP cameras.""" + + @staticmethod + @callback + def async_get_options_flow( + config_entry: VivotekConfigEntry, + ) -> OptionsFlowHandler: + """Create the options flow.""" + return OptionsFlowHandler() + + 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: + self._async_abort_entries_match( + {CONF_IP_ADDRESS: user_input[CONF_IP_ADDRESS]} + ) + try: + cam_client = build_cam_client(user_input) + mac_address = await self.hass.async_add_executor_job(cam_client.get_mac) + except VivotekCameraError: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Unexpected error during camera connection test") + errors["base"] = "unknown" + else: + # prevent duplicates + await self.async_set_unique_id(format_mac(mac_address)) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=DEFAULT_NAME, + data=user_input, + options={CONF_FRAMERATE: DEFAULT_FRAMERATE}, + ) + + return self.async_show_form( + step_id="user", + data_schema=self.add_suggested_values_to_schema( + CONF_SCHEMA, user_input or {} + ), + errors=errors, + description_placeholders=DESCRIPTION_PLACEHOLDERS, + ) + + async def async_step_import( + self, import_data: (dict[str, Any]) + ) -> ConfigFlowResult: + """Import a Yaml config.""" + self._async_abort_entries_match({CONF_IP_ADDRESS: import_data[CONF_IP_ADDRESS]}) + port = 443 if import_data[CONF_SSL] else 80 + try: + cam_client = build_cam_client({**import_data, CONF_PORT: port}) + mac_address = await self.hass.async_add_executor_job(cam_client.get_mac) + except VivotekCameraError: + return self.async_abort(reason="cannot_connect") + except Exception: + _LOGGER.exception("Unexpected error during camera connection test") + return self.async_abort(reason="unknown") + await self.async_set_unique_id(format_mac(mac_address)) + self._abort_if_unique_id_configured() + + return self.async_create_entry( + title=import_data.get(CONF_NAME, DEFAULT_NAME), + data={ + CONF_IP_ADDRESS: import_data[CONF_IP_ADDRESS], + CONF_PORT: port, + CONF_PASSWORD: import_data[CONF_PASSWORD], + CONF_USERNAME: import_data[CONF_USERNAME], + CONF_AUTHENTICATION: import_data[CONF_AUTHENTICATION], + CONF_SSL: import_data[CONF_SSL], + CONF_VERIFY_SSL: import_data[CONF_VERIFY_SSL], + CONF_SECURITY_LEVEL: import_data[CONF_SECURITY_LEVEL], + CONF_STREAM_PATH: import_data[CONF_STREAM_PATH], + }, + options={ + CONF_FRAMERATE: import_data[CONF_FRAMERATE], + }, + ) diff --git a/homeassistant/components/vivotek/const.py b/homeassistant/components/vivotek/const.py new file mode 100644 index 0000000000000..a674680b8cb88 --- /dev/null +++ b/homeassistant/components/vivotek/const.py @@ -0,0 +1,11 @@ +"""Constants for the Vivotek integration.""" + +CONF_FRAMERATE = "framerate" +CONF_SECURITY_LEVEL = "security_level" +CONF_STREAM_PATH = "stream_path" + +DOMAIN = "vivotek" + +MANUFACTURER = "Vivotek" + +INTEGRATION_TITLE = "Vivotek" diff --git a/homeassistant/components/vivotek/manifest.json b/homeassistant/components/vivotek/manifest.json index 74a8bf9b75047..360cf73a7a7a8 100644 --- a/homeassistant/components/vivotek/manifest.json +++ b/homeassistant/components/vivotek/manifest.json @@ -2,9 +2,9 @@ "domain": "vivotek", "name": "VIVOTEK", "codeowners": ["@HarlemSquirrel"], + "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/vivotek", "iot_class": "local_polling", "loggers": ["libpyvivotek"], - "quality_scale": "legacy", "requirements": ["libpyvivotek==0.6.1"] } diff --git a/homeassistant/components/vivotek/strings.json b/homeassistant/components/vivotek/strings.json new file mode 100644 index 0000000000000..6aac6abde29c4 --- /dev/null +++ b/homeassistant/components/vivotek/strings.json @@ -0,0 +1,48 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "step": { + "user": { + "data": { + "authentication": "Authentication", + "ip_address": "[%key:common::config_flow::data::ip%]", + "name": "[%key:common::config_flow::data::name%]", + "password": "[%key:common::config_flow::data::password%]", + "port": "[%key:common::config_flow::data::port%]", + "security_level": "Security level", + "ssl": "[%key:common::config_flow::data::ssl%]", + "stream_path": "Stream path", + "username": "[%key:common::config_flow::data::username%]", + "verify_ssl": "[%key:common::config_flow::data::verify_ssl%]" + }, + "description": "Set required parameters to connect to your camera. For more information, please refer to the [integration documentation]({doc_url})" + } + } + }, + "issues": { + "deprecated_yaml_import_issue_cannot_connect": { + "description": "Configuring Vivotek using camera platform YAML configuration is deprecated.\n\nWhile importing your configuration, Home Assistant could not connect to the device. Please review the configuration and the connection to the camera, then restart Home Assistant to try again, or remove the existing YAML configuration and set the integration up via the UI.", + "title": "Vivotek YAML configuration deprecated" + }, + "deprecated_yaml_import_issue_unknown": { + "description": "Configuring Vivotek using camera platform YAML configuration is deprecated.\n\nWhile importing your configuration, an unknown error occurred. Please restart Home Assistant to try again, or remove the existing YAML configuration and set the integration up via the UI.", + "title": "[%key:component::vivotek::issues::deprecated_yaml_import_issue_cannot_connect::title%]" + } + }, + "options": { + "step": { + "init": { + "data": { + "framerate": "Frame rate" + } + } + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 31d1a49c9105f..92ba009cad4da 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -735,6 +735,7 @@ "victron_ble", "victron_remote_monitoring", "vilfo", + "vivotek", "vizio", "vlc_telnet", "vodafone_station", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 3db109f20c15f..552f34746d248 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -7339,7 +7339,7 @@ "vivotek": { "name": "VIVOTEK", "integration_type": "hub", - "config_flow": false, + "config_flow": true, "iot_class": "local_polling" }, "vizio": { diff --git a/requirements_test_all.txt b/requirements_test_all.txt index cf5759988ee3d..7c530e1a95a87 100644 --- a/requirements_test_all.txt +++ b/requirements_test_all.txt @@ -1206,6 +1206,9 @@ letpot==0.6.3 # homeassistant.components.foscam libpyfoscamcgi==0.0.9 +# homeassistant.components.vivotek +libpyvivotek==0.6.1 + # homeassistant.components.libre_hardware_monitor librehardwaremonitor-api==1.5.0 diff --git a/tests/components/vivotek/__init__.py b/tests/components/vivotek/__init__.py new file mode 100644 index 0000000000000..575e76a24ba54 --- /dev/null +++ b/tests/components/vivotek/__init__.py @@ -0,0 +1 @@ +"""Tests for Vivotek camera component.""" diff --git a/tests/components/vivotek/conftest.py b/tests/components/vivotek/conftest.py new file mode 100644 index 0000000000000..d6c1cb718cf73 --- /dev/null +++ b/tests/components/vivotek/conftest.py @@ -0,0 +1,82 @@ +"""Fixtures for Vivotek component tests.""" + +from collections.abc import Generator +from unittest.mock import AsyncMock, patch + +import pytest + +from homeassistant.components.vivotek import CONF_SECURITY_LEVEL +from homeassistant.components.vivotek.const import ( + CONF_FRAMERATE, + CONF_STREAM_PATH, + DOMAIN, +) +from homeassistant.const import ( + CONF_AUTHENTICATION, + CONF_IP_ADDRESS, + CONF_NAME, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, + CONF_VERIFY_SSL, + HTTP_BASIC_AUTHENTICATION, +) + +from tests.common import MockConfigEntry + +TEST_DATA = { + CONF_NAME: "Test Camera", + CONF_IP_ADDRESS: "1.2.3.4", + CONF_PORT: "80", + CONF_USERNAME: "admin", + CONF_PASSWORD: "pass1234", + CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION, + CONF_SSL: False, + CONF_VERIFY_SSL: True, + "framerate": 2, + "security_level": "admin", + "stream_path": "/live.sdp", +} + + +@pytest.fixture +def mock_setup_entry() -> Generator[None]: + """Mock setting up a config entry.""" + with patch("homeassistant.components.vivotek.async_setup_entry", return_value=True): + yield + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Mock existing config entry.""" + return MockConfigEntry( + domain=DOMAIN, + data={ + CONF_IP_ADDRESS: "1.2.3.4", + CONF_PORT: 80, + CONF_USERNAME: "admin", + CONF_PASSWORD: "pass1234", + CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION, + CONF_SSL: False, + CONF_VERIFY_SSL: True, + CONF_SECURITY_LEVEL: "admin", + CONF_STREAM_PATH: "/live.sdp", + }, + options={ + CONF_FRAMERATE: 2, + }, + title="Vivotek Camera", + unique_id="11:22:33:44:55:66", + ) + + +@pytest.fixture +def mock_vivotek_camera() -> Generator[AsyncMock]: + """Mock existing config entry.""" + with patch( + "homeassistant.components.vivotek.VivotekCamera", autospec=True + ) as vivotek_camera: + instance = vivotek_camera.return_value + instance.get_mac.return_value = "11:22:33:44:55:66" + yield instance diff --git a/tests/components/vivotek/test_config_flow.py b/tests/components/vivotek/test_config_flow.py new file mode 100644 index 0000000000000..099f0c037bed5 --- /dev/null +++ b/tests/components/vivotek/test_config_flow.py @@ -0,0 +1,252 @@ +"""Tests for the Vivotek config flow.""" + +from unittest.mock import AsyncMock + +from libpyvivotek.vivotek import VivotekCameraError +import pytest + +from homeassistant.components.vivotek.camera import DEFAULT_FRAMERATE, DEFAULT_NAME +from homeassistant.components.vivotek.const import ( + CONF_FRAMERATE, + CONF_SECURITY_LEVEL, + CONF_STREAM_PATH, + DOMAIN, +) +from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER +from homeassistant.const import ( + CONF_AUTHENTICATION, + CONF_IP_ADDRESS, + CONF_PASSWORD, + CONF_PORT, + CONF_SSL, + CONF_USERNAME, + CONF_VERIFY_SSL, + HTTP_BASIC_AUTHENTICATION, +) +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from tests.common import MockConfigEntry + +USER_DATA = { + CONF_IP_ADDRESS: "1.2.3.4", + CONF_PORT: 80, + CONF_USERNAME: "admin", + CONF_PASSWORD: "pass1234", + CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION, + CONF_SSL: False, + CONF_VERIFY_SSL: True, + CONF_SECURITY_LEVEL: "admin", + CONF_STREAM_PATH: "/live.sdp", +} + +IMPORT_DATA = { + CONF_IP_ADDRESS: "1.2.3.4", + CONF_USERNAME: "admin", + CONF_PASSWORD: "pass1234", + CONF_AUTHENTICATION: HTTP_BASIC_AUTHENTICATION, + CONF_SSL: False, + CONF_VERIFY_SSL: True, + CONF_SECURITY_LEVEL: "admin", + CONF_STREAM_PATH: "/live.sdp", + CONF_FRAMERATE: DEFAULT_FRAMERATE, +} + + +async def test_full_flow( + hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_vivotek_camera: AsyncMock +) -> None: + """Test full user initiated flow.""" + 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"], USER_DATA + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_NAME + assert result["data"] == USER_DATA + assert result["options"] == {CONF_FRAMERATE: DEFAULT_FRAMERATE} + assert result["result"].unique_id == "11:22:33:44:55:66" + + +@pytest.mark.parametrize( + ("exception", "error"), + [ + (VivotekCameraError, "cannot_connect"), + (Exception, "unknown"), + ], +) +async def test_user_exceptions( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_vivotek_camera: AsyncMock, + exception: Exception, + error: str, +) -> None: + """Test user initiated flow with exceptions.""" + 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"] == {} + + mock_vivotek_camera.get_mac.side_effect = exception + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], USER_DATA + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "user" + assert result["errors"] == {"base": error} + + mock_vivotek_camera.get_mac.side_effect = None + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], USER_DATA + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_duplicate_entry( + hass: HomeAssistant, mock_config_entry: MockConfigEntry +) -> None: + """Test flow abort on duplicate entry.""" + mock_config_entry.add_to_hass(hass) + 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"], USER_DATA + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_duplicate_entry_mac( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test flow abort on duplicate MAC address.""" + mock_config_entry.add_to_hass(hass) + 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"], {**USER_DATA, CONF_IP_ADDRESS: "1.1.1.1"} + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_import_flow( + hass: HomeAssistant, mock_setup_entry: AsyncMock, mock_vivotek_camera: AsyncMock +) -> None: + """Test import initiated flow.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_DATA + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == DEFAULT_NAME + assert result["data"] == USER_DATA + assert result["options"] == {CONF_FRAMERATE: DEFAULT_FRAMERATE} + assert result["result"].unique_id == "11:22:33:44:55:66" + + +@pytest.mark.parametrize( + ("exception", "reason"), + [ + (VivotekCameraError, "cannot_connect"), + (Exception, "unknown"), + ], +) +async def test_import_flow_exceptions( + hass: HomeAssistant, + mock_setup_entry: AsyncMock, + mock_vivotek_camera: AsyncMock, + exception: Exception, + reason: str, +) -> None: + """Test import initiated flow with exceptions.""" + mock_vivotek_camera.get_mac.side_effect = exception + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_DATA + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == reason + + +async def test_import_flow_duplicate( + hass: HomeAssistant, mock_config_entry: MockConfigEntry +) -> None: + """Test import initiated flow.""" + mock_config_entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_IMPORT}, data=IMPORT_DATA + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_import_flow_duplicate_mac( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test import initiated flow.""" + mock_config_entry.add_to_hass(hass) + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": SOURCE_IMPORT}, + data={**IMPORT_DATA, CONF_IP_ADDRESS: "1.1.1.1"}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_options_flow( + hass: HomeAssistant, + mock_vivotek_camera: AsyncMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test options flow.""" + mock_config_entry.add_to_hass(hass) + result = await hass.config_entries.options.async_init(mock_config_entry.entry_id) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "init" + + result = await hass.config_entries.options.async_configure( + result["flow_id"], + { + CONF_FRAMERATE: 15, + }, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + assert mock_config_entry.options[CONF_FRAMERATE] == 15