From 134df28515b65cdc881e5087b5154d8de2c7d95e Mon Sep 17 00:00:00 2001 From: Frank <639906+syphernl@users.noreply.github.com> Date: Fri, 28 Aug 2026 21:08:41 +0200 Subject: [PATCH] Add RFID token management services to Peblar integration (#169189) Co-authored-by: Franck Nijhof --- homeassistant/components/peblar/__init__.py | 13 +- homeassistant/components/peblar/const.py | 2 + homeassistant/components/peblar/icons.json | 11 + .../components/peblar/quality_scale.yaml | 9 +- homeassistant/components/peblar/services.py | 140 +++++++++ homeassistant/components/peblar/services.yaml | 35 +++ homeassistant/components/peblar/strings.json | 47 +++ tests/components/peblar/test_services.py | 279 ++++++++++++++++++ 8 files changed, 528 insertions(+), 8 deletions(-) create mode 100644 homeassistant/components/peblar/services.py create mode 100644 homeassistant/components/peblar/services.yaml create mode 100644 tests/components/peblar/test_services.py diff --git a/homeassistant/components/peblar/__init__.py b/homeassistant/components/peblar/__init__.py index 105e2b90cc58..ab7a2040b899 100644 --- a/homeassistant/components/peblar/__init__.py +++ b/homeassistant/components/peblar/__init__.py @@ -14,8 +14,11 @@ from peblar import ( from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady +from homeassistant.helpers import config_validation as cv from homeassistant.helpers.aiohttp_client import async_create_clientsession +from homeassistant.helpers.typing import ConfigType +from .const import DOMAIN from .coordinator import ( PeblarConfigEntry, PeblarDataUpdateCoordinator, @@ -23,6 +26,9 @@ from .coordinator import ( PeblarUserConfigurationDataUpdateCoordinator, PeblarVersionDataUpdateCoordinator, ) +from .services import async_setup_services + +CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) PLATFORMS = [ Platform.BINARY_SENSOR, @@ -35,9 +41,14 @@ PLATFORMS = [ ] +async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: + """Set up the Peblar integration.""" + async_setup_services(hass) + return True + + async def async_setup_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bool: """Set up Peblar from a config entry.""" - # Set up connection to the Peblar charger peblar = Peblar( host=entry.data[CONF_HOST], diff --git a/homeassistant/components/peblar/const.py b/homeassistant/components/peblar/const.py index b4476d731f74..cc2129baa495 100644 --- a/homeassistant/components/peblar/const.py +++ b/homeassistant/components/peblar/const.py @@ -7,6 +7,8 @@ from peblar import ChargeLimiter, CPState DOMAIN: Final = "peblar" +CONF_UID: Final = "uid" + LOGGER = logging.getLogger(__package__) PEBLAR_CHARGE_LIMITER_TO_HOME_ASSISTANT = { diff --git a/homeassistant/components/peblar/icons.json b/homeassistant/components/peblar/icons.json index 573fc87dc68d..5859b3bb8764 100644 --- a/homeassistant/components/peblar/icons.json +++ b/homeassistant/components/peblar/icons.json @@ -68,5 +68,16 @@ "default": "mdi:palette" } } + }, + "services": { + "add_rfid_token": { + "service": "mdi:card-plus" + }, + "delete_rfid_token": { + "service": "mdi:card-remove" + }, + "list_rfid_tokens": { + "service": "mdi:card-account-details" + } } } diff --git a/homeassistant/components/peblar/quality_scale.yaml b/homeassistant/components/peblar/quality_scale.yaml index ac4bc10abf1a..0182cf953bbf 100644 --- a/homeassistant/components/peblar/quality_scale.yaml +++ b/homeassistant/components/peblar/quality_scale.yaml @@ -1,18 +1,13 @@ rules: # Bronze - action-setup: - status: exempt - comment: Integration does not register custom actions. + action-setup: done appropriate-polling: done brands: done common-modules: done config-flow-test-coverage: done config-flow: done dependency-transparency: done - docs-actions: - status: exempt - comment: | - This integration does not have any custom actions. + docs-actions: done docs-conditions: status: exempt comment: This integration does not have any conditions. diff --git a/homeassistant/components/peblar/services.py b/homeassistant/components/peblar/services.py new file mode 100644 index 000000000000..6f141dd773b7 --- /dev/null +++ b/homeassistant/components/peblar/services.py @@ -0,0 +1,140 @@ +"""Services for the Peblar integration.""" + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from peblar import Peblar, PeblarAuthenticationError, PeblarConnectionError, PeblarError +import voluptuous as vol + +from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_DESCRIPTION +from homeassistant.core import ( + HomeAssistant, + ServiceCall, + ServiceResponse, + SupportsResponse, + callback, +) +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError +from homeassistant.helpers.service import ( + async_get_config_entry, + async_register_admin_service, +) + +from .const import CONF_UID, DOMAIN +from .coordinator import PeblarConfigEntry + +SERVICE_ADD_RFID_TOKEN = "add_rfid_token" +SERVICE_DELETE_RFID_TOKEN = "delete_rfid_token" +SERVICE_LIST_RFID_TOKENS = "list_rfid_tokens" + +CHARGER_SCHEMA = vol.Schema({vol.Required(ATTR_CONFIG_ENTRY_ID): str}) +TOKEN_SCHEMA = CHARGER_SCHEMA.extend({vol.Required(CONF_UID): str}) +ADD_TOKEN_SCHEMA = TOKEN_SCHEMA.extend({vol.Required(CONF_DESCRIPTION): str}) + + +def _get_peblar(hass: HomeAssistant, entry_id: str) -> Peblar: + """Return the Peblar client for the charger the call targets. + + Every action here manages the standalone authorization list, which + lives on the RFID reader, so a charger without one is turned away + here instead of failing somewhere inside the charger. + """ + entry: PeblarConfigEntry = async_get_config_entry(hass, DOMAIN, entry_id) + + if not entry.runtime_data.system_information.hardware_has_rfid: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="no_rfid_hardware", + translation_placeholders={"charger": entry.title}, + ) + + return entry.runtime_data.user_configuration_coordinator.peblar + + +@asynccontextmanager +async def _handle_peblar_errors( + hass: HomeAssistant, entry_id: str +) -> AsyncIterator[None]: + """Translate Peblar library errors into Home Assistant errors.""" + try: + yield + + except PeblarAuthenticationError as error: + # Reload the config entry to trigger reauth flow + hass.config_entries.async_schedule_reload(entry_id) + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="authentication_error", + ) from error + + except PeblarConnectionError as error: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="communication_error", + translation_placeholders={"error": str(error)}, + ) from error + + except PeblarError as error: + raise HomeAssistantError( + translation_domain=DOMAIN, + translation_key="unknown_error", + translation_placeholders={"error": str(error)}, + ) from error + + +@callback +def async_setup_services(hass: HomeAssistant) -> None: + """Register RFID management services.""" + + async def _handle_list_rfid_tokens(call: ServiceCall) -> ServiceResponse: + entry_id = call.data[ATTR_CONFIG_ENTRY_ID] + peblar = _get_peblar(hass, entry_id) + async with _handle_peblar_errors(hass, entry_id): + tokens = await peblar.rfid_tokens() + return { + "tokens": [ + { + "uid": token.rfid_token_uid, + CONF_DESCRIPTION: token.rfid_token_description, + } + for token in tokens + ] + } + + async def _handle_add_rfid_token(call: ServiceCall) -> None: + entry_id = call.data[ATTR_CONFIG_ENTRY_ID] + peblar = _get_peblar(hass, entry_id) + async with _handle_peblar_errors(hass, entry_id): + await peblar.add_rfid_token( + rfid_token_uid=call.data[CONF_UID], + rfid_token_description=call.data[CONF_DESCRIPTION], + ) + + async def _handle_delete_rfid_token(call: ServiceCall) -> None: + entry_id = call.data[ATTR_CONFIG_ENTRY_ID] + peblar = _get_peblar(hass, entry_id) + async with _handle_peblar_errors(hass, entry_id): + await peblar.delete_rfid_token(uid=call.data[CONF_UID]) + + async_register_admin_service( + hass, + DOMAIN, + SERVICE_LIST_RFID_TOKENS, + _handle_list_rfid_tokens, + schema=CHARGER_SCHEMA, + supports_response=SupportsResponse.ONLY, + ) + async_register_admin_service( + hass, + DOMAIN, + SERVICE_ADD_RFID_TOKEN, + _handle_add_rfid_token, + schema=ADD_TOKEN_SCHEMA, + ) + async_register_admin_service( + hass, + DOMAIN, + SERVICE_DELETE_RFID_TOKEN, + _handle_delete_rfid_token, + schema=TOKEN_SCHEMA, + ) diff --git a/homeassistant/components/peblar/services.yaml b/homeassistant/components/peblar/services.yaml new file mode 100644 index 000000000000..179a2de9a4ed --- /dev/null +++ b/homeassistant/components/peblar/services.yaml @@ -0,0 +1,35 @@ +list_rfid_tokens: + fields: + config_entry_id: + required: true + selector: + config_entry: + integration: peblar + +add_rfid_token: + fields: + config_entry_id: + required: true + selector: + config_entry: + integration: peblar + uid: + required: true + selector: + text: + description: + required: true + selector: + text: + +delete_rfid_token: + fields: + config_entry_id: + required: true + selector: + config_entry: + integration: peblar + uid: + required: true + selector: + text: diff --git a/homeassistant/components/peblar/strings.json b/homeassistant/components/peblar/strings.json index 2c96c31130b8..e63635ada62d 100644 --- a/homeassistant/components/peblar/strings.json +++ b/homeassistant/components/peblar/strings.json @@ -195,8 +195,55 @@ "communication_error": { "message": "An error occurred while communicating with the Peblar EV charger: {error}" }, + "no_rfid_hardware": { + "message": "{charger} has no RFID reader, so it has no standalone authorization list." + }, "unknown_error": { "message": "An unknown error occurred while communicating with the Peblar EV charger: {error}" } + }, + "services": { + "add_rfid_token": { + "description": "Adds an RFID token to the charger's standalone authorization list.", + "fields": { + "config_entry_id": { + "description": "The Peblar EV charger to add the RFID token to.", + "name": "Peblar EV charger" + }, + "description": { + "description": "A human-readable label for this RFID token.", + "name": "Description" + }, + "uid": { + "description": "The unique identifier of the RFID token.", + "name": "UID" + } + }, + "name": "Add RFID token" + }, + "delete_rfid_token": { + "description": "Deletes an RFID token from the charger's standalone authorization list.", + "fields": { + "config_entry_id": { + "description": "The Peblar EV charger to delete the RFID token from.", + "name": "Peblar EV charger" + }, + "uid": { + "description": "The unique identifier of the RFID token to delete.", + "name": "UID" + } + }, + "name": "Delete RFID token" + }, + "list_rfid_tokens": { + "description": "Returns the RFID tokens configured in the charger's standalone authorization list.", + "fields": { + "config_entry_id": { + "description": "The Peblar EV charger to list RFID tokens for.", + "name": "Peblar EV charger" + } + }, + "name": "List RFID tokens" + } } } diff --git a/tests/components/peblar/test_services.py b/tests/components/peblar/test_services.py new file mode 100644 index 000000000000..94764e5d76b0 --- /dev/null +++ b/tests/components/peblar/test_services.py @@ -0,0 +1,279 @@ +"""Tests for the Peblar integration services.""" + +from typing import Any +from unittest.mock import MagicMock + +from peblar import ( + PeblarAuthenticationError, + PeblarConnectionError, + PeblarError, + PeblarRfidToken, +) +import pytest + +from homeassistant.components.peblar.const import DOMAIN +from homeassistant.components.peblar.services import ( + SERVICE_ADD_RFID_TOKEN, + SERVICE_DELETE_RFID_TOKEN, + SERVICE_LIST_RFID_TOKENS, +) +from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError + +from tests.common import MockConfigEntry + + +async def test_services_registered_on_setup( + hass: HomeAssistant, + init_integration: MockConfigEntry, +) -> None: + """Test that RFID services are registered when entry is loaded.""" + assert hass.services.has_service(DOMAIN, SERVICE_LIST_RFID_TOKENS) + assert hass.services.has_service(DOMAIN, SERVICE_ADD_RFID_TOKEN) + assert hass.services.has_service(DOMAIN, SERVICE_DELETE_RFID_TOKEN) + + +async def test_services_survive_entry_unload( + hass: HomeAssistant, + init_integration: MockConfigEntry, +) -> None: + """Test RFID services stay registered when the last Peblar entry unloads.""" + await hass.config_entries.async_unload(init_integration.entry_id) + await hass.async_block_till_done() + + assert hass.services.has_service(DOMAIN, SERVICE_LIST_RFID_TOKENS) + assert hass.services.has_service(DOMAIN, SERVICE_ADD_RFID_TOKEN) + assert hass.services.has_service(DOMAIN, SERVICE_DELETE_RFID_TOKEN) + + +async def test_list_rfid_tokens( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, +) -> None: + """Test list_rfid_tokens returns token list.""" + mock_peblar.rfid_tokens.return_value = [ + PeblarRfidToken( + rfid_token_uid="AA:BB:CC:DD", + rfid_token_description="My Card", + ), + PeblarRfidToken( + rfid_token_uid="11:22:33:44", + rfid_token_description="Work Badge", + ), + ] + + result = await hass.services.async_call( + DOMAIN, + SERVICE_LIST_RFID_TOKENS, + {"config_entry_id": init_integration.entry_id}, + blocking=True, + return_response=True, + ) + + assert result == { + "tokens": [ + {"uid": "AA:BB:CC:DD", "description": "My Card"}, + {"uid": "11:22:33:44", "description": "Work Badge"}, + ] + } + mock_peblar.rfid_tokens.assert_called_once_with() + + +async def test_add_rfid_token( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, +) -> None: + """Test add_rfid_token calls library with correct args.""" + await hass.services.async_call( + DOMAIN, + SERVICE_ADD_RFID_TOKEN, + { + "config_entry_id": init_integration.entry_id, + "uid": "AA:BB:CC:DD", + "description": "My Card", + }, + blocking=True, + ) + + mock_peblar.add_rfid_token.assert_called_once_with( + rfid_token_uid="AA:BB:CC:DD", + rfid_token_description="My Card", + ) + + +async def test_delete_rfid_token( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, +) -> None: + """Test delete_rfid_token calls library with correct args.""" + await hass.services.async_call( + DOMAIN, + SERVICE_DELETE_RFID_TOKEN, + { + "config_entry_id": init_integration.entry_id, + "uid": "AA:BB:CC:DD", + }, + blocking=True, + ) + + mock_peblar.delete_rfid_token.assert_called_once_with(uid="AA:BB:CC:DD") + + +async def test_unloaded_config_entry_raises( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, +) -> None: + """Test service raises ServiceValidationError for an unloaded entry.""" + second_entry = MockConfigEntry( + domain=DOMAIN, + data=mock_config_entry.data, + unique_id="second-charger", + ) + second_entry.add_to_hass(hass) + await hass.config_entries.async_setup(second_entry.entry_id) + await hass.async_block_till_done() + + await hass.config_entries.async_unload(init_integration.entry_id) + await hass.async_block_till_done() + + with pytest.raises(ServiceValidationError) as excinfo: + await hass.services.async_call( + DOMAIN, + SERVICE_LIST_RFID_TOKENS, + {"config_entry_id": init_integration.entry_id}, + blocking=True, + return_response=True, + ) + + assert excinfo.value.translation_key == "service_config_entry_not_loaded" + + +SERVICE_CALLS: list[tuple[str, str, dict[str, Any]]] = [ + (SERVICE_LIST_RFID_TOKENS, "rfid_tokens", {}), + ( + SERVICE_ADD_RFID_TOKEN, + "add_rfid_token", + {"uid": "AA:BB:CC:DD", "description": "My Card"}, + ), + (SERVICE_DELETE_RFID_TOKEN, "delete_rfid_token", {"uid": "AA:BB:CC:DD"}), +] + + +@pytest.mark.parametrize(("service", "method_name", "service_data"), SERVICE_CALLS) +@pytest.mark.parametrize( + ("error", "translation_key"), + [ + (PeblarConnectionError("Could not connect"), "communication_error"), + (PeblarError("Something went wrong"), "unknown_error"), + ], +) +async def test_service_communication_error( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, + service: str, + method_name: str, + service_data: dict[str, Any], + error: Exception, + translation_key: str, +) -> None: + """Test Peblar library errors are translated into Home Assistant errors.""" + getattr(mock_peblar, method_name).side_effect = error + + with pytest.raises(HomeAssistantError) as excinfo: + await hass.services.async_call( + DOMAIN, + service, + {"config_entry_id": init_integration.entry_id, **service_data}, + blocking=True, + return_response=service == "list_rfid_tokens", + ) + + assert excinfo.value.translation_domain == DOMAIN + assert excinfo.value.translation_key == translation_key + assert excinfo.value.translation_placeholders == {"error": str(error)} + + +@pytest.mark.parametrize(("service", "method_name", "service_data"), SERVICE_CALLS) +async def test_service_authentication_error( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, + service: str, + method_name: str, + service_data: dict[str, Any], +) -> None: + """Test an authentication error triggers a reauthentication flow.""" + getattr(mock_peblar, method_name).side_effect = PeblarAuthenticationError( + "Authentication error" + ) + mock_peblar.login.side_effect = PeblarAuthenticationError("Authentication error") + + with pytest.raises(HomeAssistantError) as excinfo: + await hass.services.async_call( + DOMAIN, + service, + {"config_entry_id": init_integration.entry_id, **service_data}, + blocking=True, + return_response=service == "list_rfid_tokens", + ) + + assert excinfo.value.translation_domain == DOMAIN + assert excinfo.value.translation_key == "authentication_error" + assert not excinfo.value.translation_placeholders + + await hass.async_block_till_done() + assert init_integration.state is ConfigEntryState.SETUP_ERROR + + flows = hass.config_entries.flow.async_progress() + assert len(flows) == 1 + assert flows[0]["step_id"] == "reauth_confirm" + assert flows[0]["context"].get("source") == SOURCE_REAUTH + assert flows[0]["context"].get("entry_id") == init_integration.entry_id + + +async def test_invalid_config_entry_raises( + hass: HomeAssistant, + mock_peblar: MagicMock, + init_integration: MockConfigEntry, +) -> None: + """Test service raises ServiceValidationError for unknown entry ID.""" + with pytest.raises(ServiceValidationError) as excinfo: + await hass.services.async_call( + DOMAIN, + SERVICE_LIST_RFID_TOKENS, + {"config_entry_id": "nonexistent-entry-id"}, + blocking=True, + return_response=True, + ) + + assert excinfo.value.translation_key == "service_config_entry_not_found" + + +@pytest.mark.parametrize(("service", "method_name", "service_data"), SERVICE_CALLS) +@pytest.mark.parametrize("mock_peblar", [{"HwHasRfid": False}], indirect=True) +async def test_charger_without_rfid_reader( + hass: HomeAssistant, + init_integration: MockConfigEntry, + service: str, + method_name: str, + service_data: dict[str, Any], +) -> None: + """A charger without a reader has no standalone list to manage.""" + with pytest.raises(ServiceValidationError) as excinfo: + await hass.services.async_call( + DOMAIN, + service, + {"config_entry_id": init_integration.entry_id, **service_data}, + blocking=True, + return_response=service == SERVICE_LIST_RFID_TOKENS, + ) + + assert excinfo.value.translation_domain == DOMAIN + assert excinfo.value.translation_key == "no_rfid_hardware"