diff --git a/homeassistant/components/bmw_connected_drive/notify.py b/homeassistant/components/bmw_connected_drive/notify.py index 8edde1a5cd7..56523351e66 100644 --- a/homeassistant/components/bmw_connected_drive/notify.py +++ b/homeassistant/components/bmw_connected_drive/notify.py @@ -5,32 +5,35 @@ from __future__ import annotations import logging from typing import Any, cast -from bimmer_connected.models import MyBMWAPIError +from bimmer_connected.models import MyBMWAPIError, PointOfInterest from bimmer_connected.vehicle import MyBMWVehicle +import voluptuous as vol from homeassistant.components.notify import ( ATTR_DATA, ATTR_TARGET, BaseNotificationService, ) -from homeassistant.const import ( - ATTR_LATITUDE, - ATTR_LOCATION, - ATTR_LONGITUDE, - ATTR_NAME, - CONF_ENTITY_ID, -) +from homeassistant.const import ATTR_LATITUDE, ATTR_LONGITUDE, CONF_ENTITY_ID from homeassistant.core import HomeAssistant -from homeassistant.exceptions import HomeAssistantError +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError +from homeassistant.helpers import config_validation as cv from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from . import BMWConfigEntry +from . import DOMAIN, BMWConfigEntry -ATTR_LAT = "lat" ATTR_LOCATION_ATTRIBUTES = ["street", "city", "postal_code", "country"] -ATTR_LON = "lon" -ATTR_SUBJECT = "subject" -ATTR_TEXT = "text" + +POI_SCHEMA = vol.Schema( + { + vol.Required(ATTR_LATITUDE): cv.latitude, + vol.Required(ATTR_LONGITUDE): cv.longitude, + vol.Optional("street"): cv.string, + vol.Optional("city"): cv.string, + vol.Optional("postal_code"): cv.string, + vol.Optional("country"): cv.string, + } +) _LOGGER = logging.getLogger(__name__) @@ -71,33 +74,34 @@ class BMWNotificationService(BaseNotificationService): async def async_send_message(self, message: str = "", **kwargs: Any) -> None: """Send a message or POI to the car.""" + + try: + # Verify data schema + poi_data = kwargs.get(ATTR_DATA) or {} + POI_SCHEMA(poi_data) + + # Create the POI object + poi = PointOfInterest( + lat=poi_data.pop(ATTR_LATITUDE), + lon=poi_data.pop(ATTR_LONGITUDE), + name=(message or None), + **poi_data, + ) + + except (vol.Invalid, TypeError, ValueError) as ex: + raise ServiceValidationError( + translation_domain=DOMAIN, + translation_key="invalid_poi", + translation_placeholders={ + "poi_exception": str(ex), + }, + ) from ex + for vehicle in kwargs[ATTR_TARGET]: vehicle = cast(MyBMWVehicle, vehicle) _LOGGER.debug("Sending message to %s", vehicle.name) - # Extract params from data dict - data = kwargs.get(ATTR_DATA) - - # Check if message is a POI - if data is not None and ATTR_LOCATION in data: - location_dict = { - ATTR_LAT: data[ATTR_LOCATION][ATTR_LATITUDE], - ATTR_LON: data[ATTR_LOCATION][ATTR_LONGITUDE], - ATTR_NAME: message, - } - # Update dictionary with additional attributes if available - location_dict.update( - { - k: v - for k, v in data[ATTR_LOCATION].items() - if k in ATTR_LOCATION_ATTRIBUTES - } - ) - try: - await vehicle.remote_services.trigger_send_poi(location_dict) - except TypeError as ex: - raise ValueError(str(ex)) from ex - except MyBMWAPIError as ex: - raise HomeAssistantError(ex) from ex - else: - raise ValueError(f"'data.{ATTR_LOCATION}' is required.") + try: + await vehicle.remote_services.trigger_send_poi(poi) + except MyBMWAPIError as ex: + raise HomeAssistantError(ex) from ex diff --git a/homeassistant/components/bmw_connected_drive/strings.json b/homeassistant/components/bmw_connected_drive/strings.json index e7606232411..886e7b54953 100644 --- a/homeassistant/components/bmw_connected_drive/strings.json +++ b/homeassistant/components/bmw_connected_drive/strings.json @@ -168,5 +168,10 @@ "rest_of_world": "Rest of world" } } + }, + "exceptions": { + "invalid_poi": { + "message": "Invalid data for point of interest: {poi_exception}" + } } } diff --git a/tests/components/bmw_connected_drive/__init__.py b/tests/components/bmw_connected_drive/__init__.py index c11d5ef0021..3632bfc1332 100644 --- a/tests/components/bmw_connected_drive/__init__.py +++ b/tests/components/bmw_connected_drive/__init__.py @@ -1,6 +1,10 @@ """Tests for the for the BMW Connected Drive integration.""" -from bimmer_connected.const import REMOTE_SERVICE_BASE_URL, VEHICLE_CHARGING_BASE_URL +from bimmer_connected.const import ( + REMOTE_SERVICE_BASE_URL, + VEHICLE_CHARGING_BASE_URL, + VEHICLE_POI_URL, +) import respx from homeassistant import config_entries @@ -71,6 +75,7 @@ def check_remote_service_call( or c.request.url.path.startswith( VEHICLE_CHARGING_BASE_URL.replace("/{vin}", "") ) + or c.request.url.path == VEHICLE_POI_URL ) assert ( first_remote_service_call.request.url.path.endswith(remote_service) is True @@ -87,6 +92,10 @@ def check_remote_service_call( == remote_service_params ) + # Send POI doesn't return a status response, so we can't check it + if remote_service == "send-to-car": + return + # Now check final result last_event_status_call = next( c for c in reversed(router.calls) if c.request.url.path.endswith("eventStatus") diff --git a/tests/components/bmw_connected_drive/test_notify.py b/tests/components/bmw_connected_drive/test_notify.py new file mode 100644 index 00000000000..4113f618be0 --- /dev/null +++ b/tests/components/bmw_connected_drive/test_notify.py @@ -0,0 +1,151 @@ +"""Test BMW numbers.""" + +from unittest.mock import AsyncMock + +from bimmer_connected.models import MyBMWAPIError, MyBMWRemoteServiceError +from bimmer_connected.tests.common import POI_DATA +from bimmer_connected.vehicle.remote_services import RemoteServices +import pytest +import respx + +from homeassistant.core import HomeAssistant +from homeassistant.exceptions import HomeAssistantError, ServiceValidationError + +from . import check_remote_service_call, setup_mocked_integration + + +async def test_legacy_notify_service_simple( + hass: HomeAssistant, + bmw_fixture: respx.Router, +) -> None: + """Test successful sending of POIs.""" + + # Setup component + assert await setup_mocked_integration(hass) + + # Minimal required data + await hass.services.async_call( + "notify", + "bmw_connected_drive_ix_xdrive50", + { + "message": POI_DATA.get("name"), + "data": { + "latitude": POI_DATA.get("lat"), + "longitude": POI_DATA.get("lon"), + }, + }, + blocking=True, + ) + check_remote_service_call(bmw_fixture, "send-to-car") + + bmw_fixture.reset() + + # Full data + await hass.services.async_call( + "notify", + "bmw_connected_drive_ix_xdrive50", + { + "message": POI_DATA.get("name"), + "data": { + "latitude": POI_DATA.get("lat"), + "longitude": POI_DATA.get("lon"), + "street": POI_DATA.get("street"), + "city": POI_DATA.get("city"), + "postal_code": POI_DATA.get("postal_code"), + "country": POI_DATA.get("country"), + }, + }, + blocking=True, + ) + check_remote_service_call(bmw_fixture, "send-to-car") + + +@pytest.mark.usefixtures("bmw_fixture") +@pytest.mark.parametrize( + ("data", "exc_translation"), + [ + ( + { + "latitude": POI_DATA.get("lat"), + }, + "Invalid data for point of interest: required key not provided @ data['longitude']", + ), + ( + { + "latitude": POI_DATA.get("lat"), + "longitude": "text", + }, + "Invalid data for point of interest: invalid longitude for dictionary value @ data['longitude']", + ), + ( + { + "latitude": POI_DATA.get("lat"), + "longitude": 9999, + }, + "Invalid data for point of interest: invalid longitude for dictionary value @ data['longitude']", + ), + ], +) +async def test_service_call_invalid_input( + hass: HomeAssistant, + data: dict, + exc_translation: str, +) -> None: + """Test invalid inputs.""" + + # Setup component + assert await setup_mocked_integration(hass) + + with pytest.raises(ServiceValidationError) as exc: + await hass.services.async_call( + "notify", + "bmw_connected_drive_ix_xdrive50", + { + "message": POI_DATA.get("name"), + "data": data, + }, + blocking=True, + ) + assert str(exc.value) == exc_translation + + +@pytest.mark.usefixtures("bmw_fixture") +@pytest.mark.parametrize( + ("raised", "expected"), + [ + (MyBMWRemoteServiceError, HomeAssistantError), + (MyBMWAPIError, HomeAssistantError), + ], +) +async def test_service_call_fail( + hass: HomeAssistant, + raised: Exception, + expected: Exception, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Test exception handling.""" + + # Setup component + assert await setup_mocked_integration(hass) + + # Setup exception + monkeypatch.setattr( + RemoteServices, + "trigger_remote_service", + AsyncMock(side_effect=raised), + ) + + # Test + with pytest.raises(expected): + await hass.services.async_call( + "notify", + "bmw_connected_drive_ix_xdrive50", + { + "message": POI_DATA.get("name"), + "data": { + "latitude": POI_DATA.get("lat"), + "longitude": POI_DATA.get("lon"), + }, + }, + blocking=True, + )