Add RFID token management services to Peblar integration (#169189)

Co-authored-by: Franck Nijhof <git@frenck.dev>
This commit is contained in:
Frank
2026-08-28 21:08:41 +02:00
committed by GitHub
co-authored by Franck Nijhof
parent f3994e1472
commit 134df28515
8 changed files with 528 additions and 8 deletions
+12 -1
View File
@@ -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],
+2
View File
@@ -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 = {
@@ -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"
}
}
}
@@ -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.
+140
View File
@@ -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,
)
@@ -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:
@@ -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"
}
}
}
+279
View File
@@ -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"