Authorize a Peblar charge session (#180678)

homeassistant/components/peblar/icons.json

@@ -88,6 +88,9 @@
     "add_vehicle_token": {
       "service": "mdi:car-connected"
     },
+    "authorize_charge_session": {
+      "service": "mdi:card-account-details-star"
+    },
     "delete_rfid_token": {
       "service": "mdi:card-remove"
     },

homeassistant/components/peblar/services.py

@@ -3,7 +3,13 @@
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 
-from peblar import Peblar, PeblarAuthenticationError, PeblarConnectionError, PeblarError
+from peblar import (
+    Peblar,
+    PeblarApi,
+    PeblarAuthenticationError,
+    PeblarConnectionError,
+    PeblarError,
+)
 import voluptuous as vol
 
 from homeassistant.const import ATTR_CONFIG_ENTRY_ID, CONF_ALIAS, CONF_DESCRIPTION
@@ -15,6 +21,7 @@ from homeassistant.core import (
     callback,
 )
 from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
+from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.service import (
     async_get_config_entry,
     async_register_admin_service,
@@ -24,6 +31,7 @@ from .const import CONF_EVCC_ID, CONF_UID, DOMAIN
 from .coordinator import PeblarConfigEntry
 
 SERVICE_ADD_RFID_TOKEN = "add_rfid_token"
+SERVICE_AUTHORIZE_CHARGE_SESSION = "authorize_charge_session"
 SERVICE_ADD_VEHICLE_TOKEN = "add_vehicle_token"
 SERVICE_DELETE_RFID_TOKEN = "delete_rfid_token"
 SERVICE_DELETE_VEHICLE_TOKEN = "delete_vehicle_token"
@@ -38,6 +46,18 @@ ADD_TOKEN_SCHEMA = TOKEN_SCHEMA.extend({vol.Required(CONF_DESCRIPTION): str})
 VEHICLE_SCHEMA = CHARGER_SCHEMA.extend({vol.Required(CONF_EVCC_ID): str})
 ADD_VEHICLE_SCHEMA = VEHICLE_SCHEMA.extend({vol.Required(CONF_ALIAS): str})
 
+# The charger takes the token by UID or by description, and wants exactly
+# one of the two.
+AUTHORIZE_SCHEMA = vol.All(
+    CHARGER_SCHEMA.extend(
+        {
+            vol.Exclusive(CONF_UID, "token"): str,
+            vol.Exclusive(CONF_DESCRIPTION, "token"): str,
+        }
+    ),
+    cv.has_at_least_one_key(CONF_UID, CONF_DESCRIPTION),
+)
+
 
 def _get_rfid_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
     """Return the client, for a charger that has an RFID reader.
@@ -58,6 +78,47 @@ def _get_rfid_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
     return entry.runtime_data.user_configuration_coordinator.peblar
 
 
+def _get_authorizing_api(hass: HomeAssistant, entry_id: str) -> PeblarApi:
+    """Return the local REST API, for a charger that authorizes sessions.
+
+    Presenting a token lives on the local REST API rather than the web
+    one, unlike the actions that manage the lists it draws from.
+
+    The token comes from the standalone authorization list, so the reader
+    has to be there. Beyond that there are two ways for this to be
+    pointless: a charger managed by a backoffice over OCPP decides for
+    itself and refuses the request, and a charger set to charge without
+    authorization has nothing to authorize in the first place.
+    """
+    entry: PeblarConfigEntry = async_get_config_entry(hass, DOMAIN, entry_id)
+    runtime_data = entry.runtime_data
+
+    if not runtime_data.system_information.hardware_has_rfid:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="no_rfid_hardware",
+            translation_placeholders={"charger": entry.title},
+        )
+
+    configuration = runtime_data.user_configuration_coordinator.data
+
+    if configuration.secc_ocpp_active:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="managed_by_backoffice",
+            translation_placeholders={"charger": entry.title},
+        )
+
+    if configuration.session_manager_charge_without_authentication:
+        raise ServiceValidationError(
+            translation_domain=DOMAIN,
+            translation_key="authorization_not_required",
+            translation_placeholders={"charger": entry.title},
+        )
+
+    return runtime_data.data_coordinator.api
+
+
 def _get_autocharge_peblar(hass: HomeAssistant, entry_id: str) -> Peblar:
     """Return the client, for a charger that can do autocharge.
 
@@ -143,6 +204,15 @@ def async_setup_services(hass: HomeAssistant) -> None:
         async with _handle_peblar_errors(hass, entry_id):
             await peblar.delete_rfid_token(uid=call.data[CONF_UID])
 
+    async def _handle_authorize_charge_session(call: ServiceCall) -> None:
+        entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
+        api = _get_authorizing_api(hass, entry_id)
+        async with _handle_peblar_errors(hass, entry_id):
+            await api.authorize_charge_session(
+                token=call.data.get(CONF_UID),
+                name=call.data.get(CONF_DESCRIPTION),
+            )
+
     async def _handle_list_vehicle_tokens(call: ServiceCall) -> ServiceResponse:
         entry_id = call.data[ATTR_CONFIG_ENTRY_ID]
         peblar = _get_autocharge_peblar(hass, entry_id)
@@ -214,3 +284,10 @@ def async_setup_services(hass: HomeAssistant) -> None:
         _handle_delete_vehicle_token,
         schema=VEHICLE_SCHEMA,
     )
+    async_register_admin_service(
+        hass,
+        DOMAIN,
+        SERVICE_AUTHORIZE_CHARGE_SESSION,
+        _handle_authorize_charge_session,
+        schema=AUTHORIZE_SCHEMA,
+    )

homeassistant/components/peblar/services.yaml

@@ -69,3 +69,17 @@ delete_vehicle_token:
       required: true
       selector:
         text:
+
+authorize_charge_session:
+  fields:
+    config_entry_id:
+      required: true
+      selector:
+        config_entry:
+          integration: peblar
+    uid:
+      selector:
+        text:
+    description:
+      selector:
+        text:

homeassistant/components/peblar/strings.json

@@ -203,9 +203,15 @@
     "authentication_error": {
       "message": "An authentication failure occurred while communicating with the Peblar EV charger."
     },
+    "authorization_not_required": {
+      "message": "{charger} is set to charge without authorization, so there is nothing to authorize."
+    },
     "communication_error": {
       "message": "An error occurred while communicating with the Peblar EV charger: {error}"
     },
+    "managed_by_backoffice": {
+      "message": "{charger} is managed over OCPP, so its sessions are authorized by the backoffice."
+    },
     "no_autocharge_hardware": {
       "message": "{charger} has no power line communication hardware, so it cannot use autocharge."
     },
@@ -253,6 +259,24 @@
       },
       "name": "Add autocharge vehicle"
     },
+    "authorize_charge_session": {
+      "description": "Presents a token to the charger, as if it were held against the reader. This authorizes a session that is waiting for it, and stops one that is already running.",
+      "fields": {
+        "config_entry_id": {
+          "description": "The Peblar EV charger to present the token to.",
+          "name": "Peblar EV charger"
+        },
+        "description": {
+          "description": "The label of the RFID token to present, instead of its UID.",
+          "name": "Description"
+        },
+        "uid": {
+          "description": "The unique identifier of the RFID token to present.",
+          "name": "UID"
+        }
+      },
+      "name": "Authorize charge session"
+    },
     "delete_rfid_token": {
       "description": "Deletes an RFID token from the charger's standalone authorization list.",
       "fields": {

tests/components/peblar/test_services.py

@@ -11,11 +11,13 @@ from peblar import (
     PeblarVehicleToken,
 )
 import pytest
+import voluptuous as vol
 
 from homeassistant.components.peblar.const import DOMAIN
 from homeassistant.components.peblar.services import (
     SERVICE_ADD_RFID_TOKEN,
     SERVICE_ADD_VEHICLE_TOKEN,
+    SERVICE_AUTHORIZE_CHARGE_SESSION,
     SERVICE_DELETE_RFID_TOKEN,
     SERVICE_DELETE_VEHICLE_TOKEN,
     SERVICE_LIST_RFID_TOKENS,
@@ -409,3 +411,94 @@ async def test_autocharge_needs_power_line_communication(
 
     assert excinfo.value.translation_domain == DOMAIN
     assert excinfo.value.translation_key == "no_autocharge_hardware"
+
+
+@pytest.mark.parametrize(
+    ("service_data", "expected"),
+    [
+        ({"uid": "0123456789ABCD"}, {"token": "0123456789ABCD", "name": None}),
+        ({"description": "My card"}, {"token": None, "name": "My card"}),
+    ],
+    ids=["by uid", "by description"],
+)
+async def test_authorize_charge_session(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    init_integration: MockConfigEntry,
+    service_data: dict[str, Any],
+    expected: dict[str, Any],
+) -> None:
+    """Test the token can be presented by either of the two names for it."""
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_AUTHORIZE_CHARGE_SESSION,
+        {"config_entry_id": init_integration.entry_id, **service_data},
+        blocking=True,
+    )
+
+    mock_peblar.rest_api.return_value.authorize_charge_session.assert_called_once_with(
+        **expected
+    )
+
+
+@pytest.mark.parametrize(
+    "service_data",
+    [
+        {},
+        {"uid": "0123456789ABCD", "description": "My card"},
+    ],
+    ids=["neither", "both"],
+)
+async def test_authorize_charge_session_needs_exactly_one_token(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    init_integration: MockConfigEntry,
+    service_data: dict[str, Any],
+) -> None:
+    """Test the charger is told which token to present, and only one."""
+    with pytest.raises(vol.Invalid):
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_AUTHORIZE_CHARGE_SESSION,
+            {"config_entry_id": init_integration.entry_id, **service_data},
+            blocking=True,
+        )
+
+    mock_peblar.rest_api.return_value.authorize_charge_session.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("mock_peblar", "translation_key"),
+    [
+        ({"HwHasRfid": False}, "no_rfid_hardware"),
+        ({"SeccOcppActive": True}, "managed_by_backoffice"),
+        ({"SessionManagerChargeWithoutAuth": True}, "authorization_not_required"),
+    ],
+    ids=["no reader", "managed over OCPP", "no authorization needed"],
+    indirect=["mock_peblar"],
+)
+async def test_authorize_charge_session_is_refused(
+    hass: HomeAssistant,
+    mock_peblar: MagicMock,
+    init_integration: MockConfigEntry,
+    translation_key: str,
+) -> None:
+    """Test a charger that cannot or need not authorize is turned away.
+
+    The API refuses this outright on a charger managed over OCPP, and a
+    charger that charges without authorization has nothing to authorize.
+    """
+    with pytest.raises(ServiceValidationError) as excinfo:
+        await hass.services.async_call(
+            DOMAIN,
+            SERVICE_AUTHORIZE_CHARGE_SESSION,
+            {
+                "config_entry_id": init_integration.entry_id,
+                "uid": "0123456789ABCD",
+            },
+            blocking=True,
+        )
+
+    assert excinfo.value.translation_domain == DOMAIN
+    assert excinfo.value.translation_key == translation_key
+    mock_peblar.rest_api.return_value.authorize_charge_session.assert_not_called()