diff --git a/homeassistant/components/swisscom/__init__.py b/homeassistant/components/swisscom/__init__.py index 5e0c11af090c3e..bb277c1e7c2b42 100644 --- a/homeassistant/components/swisscom/__init__.py +++ b/homeassistant/components/swisscom/__init__.py @@ -1 +1,23 @@ -"""The swisscom component.""" +"""The Swisscom Internet-Box integration.""" + +from homeassistant.const import Platform +from homeassistant.core import HomeAssistant + +from .coordinator import SwisscomConfigEntry, SwisscomDataUpdateCoordinator + +PLATFORMS: list[Platform] = [Platform.DEVICE_TRACKER] + + +async def async_setup_entry(hass: HomeAssistant, entry: SwisscomConfigEntry) -> bool: + """Set up Swisscom Internet-Box from a config entry.""" + coordinator = SwisscomDataUpdateCoordinator(hass, entry) + await coordinator.async_config_entry_first_refresh() + entry.runtime_data = coordinator + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + return True + + +async def async_unload_entry(hass: HomeAssistant, entry: SwisscomConfigEntry) -> bool: + """Unload a config entry.""" + return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) diff --git a/homeassistant/components/swisscom/config_flow.py b/homeassistant/components/swisscom/config_flow.py new file mode 100644 index 00000000000000..45ee3848e51380 --- /dev/null +++ b/homeassistant/components/swisscom/config_flow.py @@ -0,0 +1,67 @@ +"""Config flow for the Swisscom Internet-Box integration.""" + +import logging +from typing import Any + +from swisscom_internet_box import ( + SwisscomAuthError, + SwisscomClient, + SwisscomConnectionError, +) +import voluptuous as vol + +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.device_registry import format_mac + +from .const import DEFAULT_HOST, DEFAULT_USERNAME, DOMAIN + +_LOGGER = logging.getLogger(__name__) + +STEP_USER_DATA_SCHEMA = vol.Schema( + { + vol.Required(CONF_HOST, default=DEFAULT_HOST): str, + vol.Required(CONF_USERNAME, default=DEFAULT_USERNAME): str, + vol.Required(CONF_PASSWORD): str, + } +) + + +class SwisscomConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Swisscom Internet-Box.""" + + VERSION = 1 + + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: + """Handle the user step.""" + errors: dict[str, str] = {} + if user_input is not None: + client = SwisscomClient( + async_get_clientsession(self.hass), + user_input[CONF_HOST], + user_input[CONF_USERNAME], + user_input[CONF_PASSWORD], + ) + try: + await client.login() + info = await client.get_box_info() + except SwisscomAuthError: + errors["base"] = "invalid_auth" + except SwisscomConnectionError: + errors["base"] = "cannot_connect" + except Exception: + _LOGGER.exception("Unexpected exception during Swisscom config flow") + errors["base"] = "unknown" + else: + await self.async_set_unique_id(format_mac(info.base_mac)) + self._abort_if_unique_id_configured() + return self.async_create_entry( + title=info.model_name or "Internet-Box", data=user_input + ) + + return self.async_show_form( + step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors + ) diff --git a/homeassistant/components/swisscom/const.py b/homeassistant/components/swisscom/const.py new file mode 100644 index 00000000000000..9bc400a268e812 --- /dev/null +++ b/homeassistant/components/swisscom/const.py @@ -0,0 +1,6 @@ +"""Constants for the Swisscom Internet-Box integration.""" + +DOMAIN = "swisscom" + +DEFAULT_HOST = "192.168.1.1" +DEFAULT_USERNAME = "admin" diff --git a/homeassistant/components/swisscom/coordinator.py b/homeassistant/components/swisscom/coordinator.py new file mode 100644 index 00000000000000..c59f24dc05e4f3 --- /dev/null +++ b/homeassistant/components/swisscom/coordinator.py @@ -0,0 +1,59 @@ +"""DataUpdateCoordinator for the Swisscom Internet-Box.""" + +from datetime import timedelta +import logging + +from swisscom_internet_box import ( + Device, + SwisscomAuthError, + SwisscomClient, + SwisscomConnectionError, +) + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import ConfigEntryAuthFailed +from homeassistant.helpers.aiohttp_client import async_get_clientsession +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + +SCAN_INTERVAL = timedelta(seconds=30) + +type SwisscomConfigEntry = ConfigEntry[SwisscomDataUpdateCoordinator] + + +class SwisscomDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Device]]): + """Poll the Internet-Box for the list of LAN devices.""" + + config_entry: SwisscomConfigEntry + + def __init__(self, hass: HomeAssistant, entry: SwisscomConfigEntry) -> None: + """Initialize the coordinator.""" + super().__init__( + hass, + _LOGGER, + name=DOMAIN, + update_interval=SCAN_INTERVAL, + config_entry=entry, + ) + self.client = SwisscomClient( + async_get_clientsession(hass), + entry.data[CONF_HOST], + entry.data[CONF_USERNAME], + entry.data[CONF_PASSWORD], + ) + + async def _async_update_data(self) -> dict[str, Device]: + """Fetch device data from the box.""" + try: + devices = await self.client.get_devices() + except SwisscomAuthError as err: + raise ConfigEntryAuthFailed(str(err)) from err + except SwisscomConnectionError as err: + raise UpdateFailed(str(err)) from err + + return {device.key: device for device in devices if device.key} diff --git a/homeassistant/components/swisscom/device_tracker.py b/homeassistant/components/swisscom/device_tracker.py index 902449c9f86474..b8acdb1acd8984 100644 --- a/homeassistant/components/swisscom/device_tracker.py +++ b/homeassistant/components/swisscom/device_tracker.py @@ -1,111 +1,115 @@ -"""Support for Swisscom routers (Internet-Box).""" +"""Device tracker for the Swisscom Internet-Box.""" -from contextlib import suppress -import logging - -import requests import voluptuous as vol from homeassistant.components.device_tracker import ( - DOMAIN as DEVICE_TRACKER_DOMAIN, PLATFORM_SCHEMA as DEVICE_TRACKER_PLATFORM_SCHEMA, - DeviceScanner, + AsyncSeeCallback, + ScannerEntity, ) from homeassistant.const import CONF_HOST -from homeassistant.core import HomeAssistant -from homeassistant.helpers import config_validation as cv -from homeassistant.helpers.typing import ConfigType - -_LOGGER = logging.getLogger(__name__) +from homeassistant.core import HomeAssistant, callback +from homeassistant.helpers import config_validation as cv, issue_registry as ir +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType +from homeassistant.helpers.update_coordinator import CoordinatorEntity -DEFAULT_IP = "192.168.1.1" +from .const import DEFAULT_HOST, DOMAIN +from .coordinator import SwisscomConfigEntry, SwisscomDataUpdateCoordinator PLATFORM_SCHEMA = DEVICE_TRACKER_PLATFORM_SCHEMA.extend( - {vol.Optional(CONF_HOST, default=DEFAULT_IP): cv.string} + {vol.Optional(CONF_HOST, default=DEFAULT_HOST): cv.string} ) -def get_scanner( - hass: HomeAssistant, config: ConfigType -) -> SwisscomDeviceScanner | None: - """Return the Swisscom device scanner.""" - scanner = SwisscomDeviceScanner(config[DEVICE_TRACKER_DOMAIN]) - - return scanner if scanner.success_init else None - - -class SwisscomDeviceScanner(DeviceScanner): - """Class which queries a router running Swisscom Internet-Box firmware.""" - - def __init__(self, config): - """Initialize the scanner.""" - self.host = config[CONF_HOST] - self.last_results = {} - - # Test the router is accessible. - data = self.get_swisscom_data() - self.success_init = data is not None - - def scan_devices(self): - """Scan for new devices and return a list with found device IDs.""" - self._update_info() - return [client["mac"] for client in self.last_results] - - def get_device_name(self, device): - """Return the name of the given device or None if we don't know.""" - if not self.last_results: - return None - for client in self.last_results: - if client["mac"] == device: - return client["host"] - return None - - def _update_info(self): - """Ensure the information from the Swisscom router is up to date. - - Return boolean if scanning successful. - """ - if not self.success_init: - return False - - _LOGGER.debug("Loading data from Swisscom Internet Box") - if not (data := self.get_swisscom_data()): - return False - - active_clients = [client for client in data.values() if client["status"]] - self.last_results = active_clients - return True - - def get_swisscom_data(self): - """Retrieve data from Swisscom and return parsed result.""" - url = f"http://{self.host}/ws" - headers = {"Content-Type": "application/x-sah-ws-4-call+json"} - data = """ - {"service":"Devices", "method":"get", - "parameters":{"expression":"lan and not self"}}""" - - devices = {} - - try: - request = requests.post(url, headers=headers, data=data, timeout=10) - except ( - requests.exceptions.ConnectionError, - requests.exceptions.Timeout, - requests.exceptions.ConnectTimeout, - ): - _LOGGER.debug("No response from Swisscom Internet Box") - return devices - - if "status" not in request.json(): - _LOGGER.debug("No status in response from Swisscom Internet Box") - return devices - - for device in request.json()["status"]: - with suppress(KeyError, requests.exceptions.RequestException): - devices[device["Key"]] = { - "ip": device["IPAddress"], - "mac": device["PhysAddress"], - "host": device["Name"], - "status": device["Active"], - } - return devices +async def async_setup_scanner( + hass: HomeAssistant, + config: ConfigType, + async_see: AsyncSeeCallback, + discovery_info: DiscoveryInfoType | None = None, +) -> bool: + """Inform users that the YAML configuration is no longer supported.""" + ir.async_create_issue( + hass, + DOMAIN, + "deprecated_yaml_import_issue_credentials_required", + breaks_in_ha_version="2027.1.0", + is_fixable=False, + is_persistent=False, + issue_domain=DOMAIN, + severity=ir.IssueSeverity.WARNING, + translation_key="deprecated_yaml_import_issue_credentials_required", + translation_placeholders={ + "domain": DOMAIN, + "integration_title": "Swisscom Internet-Box", + "host": config[CONF_HOST], + }, + ) + return False + + +async def async_setup_entry( + hass: HomeAssistant, + entry: SwisscomConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up device tracker entities for the Swisscom Internet-Box.""" + coordinator = entry.runtime_data + tracked: set[str] = set() + + @callback + def _add_new_entities() -> None: + new_keys = [key for key in coordinator.data if key not in tracked] + if new_keys: + tracked.update(new_keys) + async_add_entities( + SwisscomScannerEntity(coordinator, key) for key in new_keys + ) + + _add_new_entities() + entry.async_on_unload(coordinator.async_add_listener(_add_new_entities)) + + +class SwisscomScannerEntity( + CoordinatorEntity[SwisscomDataUpdateCoordinator], ScannerEntity +): + """A device tracked by the Swisscom Internet-Box.""" + + def __init__(self, coordinator: SwisscomDataUpdateCoordinator, key: str) -> None: + """Initialize the scanner entity.""" + super().__init__(coordinator) + self._key = key + self._attr_unique_id = key + + @property + def _device(self): + return self.coordinator.data.get(self._key) + + @property + def is_connected(self) -> bool: + """Return whether the device is currently connected to the LAN.""" + device = self._device + return bool(device and device.active) + + @property + def mac_address(self) -> str: + """Return the MAC address of the device.""" + device = self._device + return device.phys_address if device else self._key + + @property + def hostname(self) -> str | None: + """Return the hostname of the device.""" + device = self._device + return device.name if device else None + + @property + def ip_address(self) -> str | None: + """Return the IP address of the device.""" + device = self._device + return device.ip_address if device else None + + @property + def name(self) -> str | None: + """Return the friendly name of the device.""" + return self.hostname diff --git a/homeassistant/components/swisscom/manifest.json b/homeassistant/components/swisscom/manifest.json index cf1ea01ea9c04a..8b259e82d90daf 100644 --- a/homeassistant/components/swisscom/manifest.json +++ b/homeassistant/components/swisscom/manifest.json @@ -2,7 +2,9 @@ "domain": "swisscom", "name": "Swisscom Internet-Box", "codeowners": [], + "config_flow": true, "documentation": "https://www.home-assistant.io/integrations/swisscom", + "integration_type": "hub", "iot_class": "local_polling", - "quality_scale": "legacy" + "requirements": ["python-swisscom-internet-box==0.1.1"] } diff --git a/homeassistant/components/swisscom/strings.json b/homeassistant/components/swisscom/strings.json new file mode 100644 index 00000000000000..5ea96118dd4da8 --- /dev/null +++ b/homeassistant/components/swisscom/strings.json @@ -0,0 +1,32 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "unknown": "[%key:common::config_flow::error::unknown%]" + }, + "step": { + "user": { + "data": { + "host": "[%key:common::config_flow::data::host%]", + "password": "[%key:common::config_flow::data::password%]", + "username": "[%key:common::config_flow::data::username%]" + }, + "data_description": { + "host": "The hostname or IP address of your Swisscom Internet-Box.", + "password": "The administrator password printed on the bottom of the box.", + "username": "The administrator username, normally \"admin\"." + } + } + } + }, + "issues": { + "deprecated_yaml_import_issue_credentials_required": { + "description": "Configuring the {integration_title} integration through YAML is deprecated. The integration now requires a username and password to authenticate to your Internet-Box, which cannot be safely carried over from YAML.\n\nSet up the integration through the UI to provide your credentials (your existing host `{host}` will need to be re-entered), then remove the `{domain}` entry from your `configuration.yaml` file and restart Home Assistant.", + "title": "The {integration_title} YAML configuration is being removed" + } + } +} diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 9f9fd09fbf342a..2495b990e34453 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -727,6 +727,7 @@ "sunweg", "surepetcare", "swiss_public_transport", + "swisscom", "switchbee", "switchbot", "switchbot_cloud", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 882682864a4758..b88c3ebdc897e7 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -6903,7 +6903,7 @@ "swisscom": { "name": "Swisscom Internet-Box", "integration_type": "hub", - "config_flow": false, + "config_flow": true, "iot_class": "local_polling" }, "switchbee": { diff --git a/requirements_all.txt b/requirements_all.txt index 2fd2226bcde822..224a69fb64c268 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -2738,6 +2738,9 @@ python-snoo==0.8.3 # homeassistant.components.songpal python-songpal==0.16.2 +# homeassistant.components.swisscom +python-swisscom-internet-box==0.1.1 + # homeassistant.components.tado python-tado==0.18.16 diff --git a/tests/components/swisscom/__init__.py b/tests/components/swisscom/__init__.py new file mode 100644 index 00000000000000..1d6f0cba9aa3f2 --- /dev/null +++ b/tests/components/swisscom/__init__.py @@ -0,0 +1 @@ +"""Tests for the Swisscom Internet-Box integration.""" diff --git a/tests/components/swisscom/conftest.py b/tests/components/swisscom/conftest.py new file mode 100644 index 00000000000000..c11d731f32eb0e --- /dev/null +++ b/tests/components/swisscom/conftest.py @@ -0,0 +1,49 @@ +"""Fixtures for the Swisscom Internet-Box integration tests.""" + +from collections.abc import Generator +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from homeassistant.components.swisscom.const import DOMAIN + +from .const import TEST_BASE_MAC, TEST_FORMATTED_MAC, TEST_MODEL_NAME, USER_INPUT + +from tests.common import MockConfigEntry + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Return the default mocked config entry.""" + return MockConfigEntry( + title=TEST_MODEL_NAME, + domain=DOMAIN, + data=USER_INPUT, + unique_id=TEST_FORMATTED_MAC, + ) + + +@pytest.fixture +def mock_setup_entry() -> Generator[AsyncMock]: + """Mock setting up a config entry.""" + with patch( + "homeassistant.components.swisscom.async_setup_entry", return_value=True + ) as mock_fn: + yield mock_fn + + +@pytest.fixture +def mock_swisscom_client() -> Generator[MagicMock]: + """Mock the SwisscomClient used in the config flow.""" + box_info = MagicMock() + box_info.base_mac = TEST_BASE_MAC + box_info.model_name = TEST_MODEL_NAME + + with patch( + "homeassistant.components.swisscom.config_flow.SwisscomClient", + autospec=True, + ) as mock_cls: + client = mock_cls.return_value + client.login = AsyncMock() + client.get_box_info = AsyncMock(return_value=box_info) + yield client diff --git a/tests/components/swisscom/const.py b/tests/components/swisscom/const.py new file mode 100644 index 00000000000000..2f23967f7f05da --- /dev/null +++ b/tests/components/swisscom/const.py @@ -0,0 +1,16 @@ +"""Constants for the Swisscom Internet-Box integration tests.""" + +from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME + +TEST_HOST = "192.168.1.1" +TEST_USERNAME = "admin" +TEST_PASSWORD = "test-password" +TEST_BASE_MAC = "AA:BB:CC:DD:EE:FF" +TEST_FORMATTED_MAC = "aa:bb:cc:dd:ee:ff" +TEST_MODEL_NAME = "Internet-Box plus" + +USER_INPUT = { + CONF_HOST: TEST_HOST, + CONF_USERNAME: TEST_USERNAME, + CONF_PASSWORD: TEST_PASSWORD, +} diff --git a/tests/components/swisscom/test_config_flow.py b/tests/components/swisscom/test_config_flow.py new file mode 100644 index 00000000000000..c2152784610e59 --- /dev/null +++ b/tests/components/swisscom/test_config_flow.py @@ -0,0 +1,112 @@ +"""Tests for the Swisscom Internet-Box config flow.""" + +from unittest.mock import MagicMock + +import pytest +from swisscom_internet_box import SwisscomAuthError, SwisscomConnectionError + +from homeassistant.components.swisscom.const import DOMAIN +from homeassistant.config_entries import SOURCE_USER +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from .const import TEST_FORMATTED_MAC, TEST_MODEL_NAME, USER_INPUT + +from tests.common import MockConfigEntry + + +@pytest.mark.usefixtures("mock_setup_entry") +async def test_user_flow_success( + hass: HomeAssistant, mock_swisscom_client: MagicMock +) -> None: + """Test a successful user-initiated config 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_input=USER_INPUT + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == TEST_MODEL_NAME + assert result["data"] == USER_INPUT + assert result["result"].unique_id == TEST_FORMATTED_MAC + + +@pytest.mark.parametrize( + ("side_effect", "expected_error"), + [ + (SwisscomAuthError("bad creds"), "invalid_auth"), + (SwisscomConnectionError("unreachable"), "cannot_connect"), + (RuntimeError("boom"), "unknown"), + ], + ids=["invalid_auth", "cannot_connect", "unknown"], +) +@pytest.mark.usefixtures("mock_setup_entry") +async def test_user_flow_error_and_recovery( + hass: HomeAssistant, + mock_swisscom_client: MagicMock, + side_effect: Exception, + expected_error: str, +) -> None: + """Test user flow shows the correct error and the user can retry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + mock_swisscom_client.login.side_effect = side_effect + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=USER_INPUT + ) + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": expected_error} + + mock_swisscom_client.login.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=USER_INPUT + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == TEST_MODEL_NAME + assert result["data"] == USER_INPUT + assert result["result"].unique_id == TEST_FORMATTED_MAC + + +async def test_user_flow_duplicate( + hass: HomeAssistant, + mock_swisscom_client: MagicMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Test that duplicate boxes are rejected.""" + 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"], user_input=USER_INPUT + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +@pytest.mark.usefixtures("mock_setup_entry") +async def test_user_flow_no_model_name_uses_default_title( + hass: HomeAssistant, mock_swisscom_client: MagicMock +) -> None: + """Test the entry falls back to a default title when the box reports no model.""" + mock_swisscom_client.get_box_info.return_value.model_name = "" + + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], user_input=USER_INPUT + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "Internet-Box"