evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor BMW send_message and add tests (#119170)

Browse source 
Co-authored-by: Richard <rikroe@users.noreply.github.com>
This commit is contained in:
Richard Kroegel 2024-07-06 11:00:00 +02:00 • committed by GitHub
parent 294c31f6f5
commit f29094f41d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 210 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

84
homeassistant/components/bmw_connected_drive/notify.py
View file

@ -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

5
homeassistant/components/bmw_connected_drive/strings.json
View file

@ -168,5 +168,10 @@
"rest_of_world": "Rest of world"
}
}
},
"exceptions": {
"invalid_poi": {
"message": "Invalid data for point of interest: {poi_exception}"
}
}
}

11
tests/components/bmw_connected_drive/__init__.py
View file

@ -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")

151
tests/components/bmw_connected_drive/test_notify.py Normal file
View file

@ -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,
)