Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add Apple WeatherKit integration (#99895)

pull/99256/head
TJ Horner 2023-09-11 10:06:55 -07:00 committed by GitHub
parent 0fe88d60ac
commit 17db20fdd7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
21 changed files with 11431 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CODEOWNERS
View File

@ -1404,6 +1404,8 @@ build.json @home-assistant/supervisor
/tests/components/waze_travel_time/ @eifinger
/homeassistant/components/weather/ @home-assistant/core
/tests/components/weather/ @home-assistant/core
/homeassistant/components/weatherkit/ @tjhorner
/tests/components/weatherkit/ @tjhorner
/homeassistant/components/webhook/ @home-assistant/core
/tests/components/webhook/ @home-assistant/core
/homeassistant/components/webostv/ @thecode

3
homeassistant/brands/apple.json
View File

@ -7,6 +7,7 @@
"homekit",
"ibeacon",
"icloud",
"itunes"
"itunes",
"weatherkit"
]
}

62
homeassistant/components/weatherkit/__init__.py Normal file
View File

@ -0,0 +1,62 @@
"""Integration for Apple's WeatherKit API."""
from __future__ import annotations
from apple_weatherkit.client import (
WeatherKitApiClient,
WeatherKitApiClientAuthenticationError,
WeatherKitApiClientError,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import (
CONF_KEY_ID,
CONF_KEY_PEM,
CONF_SERVICE_ID,
CONF_TEAM_ID,
DOMAIN,
LOGGER,
)
from .coordinator import WeatherKitDataUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.WEATHER]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up this integration using UI."""
hass.data.setdefault(DOMAIN, {})
coordinator = WeatherKitDataUpdateCoordinator(
hass=hass,
client=WeatherKitApiClient(
key_id=entry.data[CONF_KEY_ID],
service_id=entry.data[CONF_SERVICE_ID],
team_id=entry.data[CONF_TEAM_ID],
key_pem=entry.data[CONF_KEY_PEM],
session=async_get_clientsession(hass),
),
)
try:
await coordinator.update_supported_data_sets()
except WeatherKitApiClientAuthenticationError as ex:
LOGGER.error("Authentication error initializing integration: %s", ex)
return False
except WeatherKitApiClientError as ex:
raise ConfigEntryNotReady from ex
await coordinator.async_config_entry_first_refresh()
hass.data[DOMAIN][entry.entry_id] = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Handle removal of an entry."""
if unloaded := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unloaded

126
homeassistant/components/weatherkit/config_flow.py Normal file
View File

@ -0,0 +1,126 @@
"""Adds config flow for WeatherKit."""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from apple_weatherkit.client import (
WeatherKitApiClient,
WeatherKitApiClientAuthenticationError,
WeatherKitApiClientCommunicationError,
WeatherKitApiClientError,
)
import voluptuous as vol
from homeassistant import config_entries, data_entry_flow
from homeassistant.const import CONF_LATITUDE, CONF_LOCATION, CONF_LONGITUDE
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import (
LocationSelector,
LocationSelectorConfig,
TextSelector,
TextSelectorConfig,
)
from .const import (
CONF_KEY_ID,
CONF_KEY_PEM,
CONF_SERVICE_ID,
CONF_TEAM_ID,
DOMAIN,
LOGGER,
)
DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_LOCATION): LocationSelector(
LocationSelectorConfig(radius=False, icon="")
),
# Auth
vol.Required(CONF_KEY_ID): str,
vol.Required(CONF_SERVICE_ID): str,
vol.Required(CONF_TEAM_ID): str,
vol.Required(CONF_KEY_PEM): TextSelector(
TextSelectorConfig(
multiline=True,
)
),
}
)
class WeatherKitUnsupportedLocationError(Exception):
"""Error to indicate a location is unsupported."""
class WeatherKitFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Config flow for WeatherKit."""
VERSION = 1
async def async_step_user(
self,
user_input: dict[str, Any] | None = None,
) -> data_entry_flow.FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
try:
await self._test_config(user_input)
except WeatherKitUnsupportedLocationError as exception:
LOGGER.error(exception)
errors["base"] = "unsupported_location"
except WeatherKitApiClientAuthenticationError as exception:
LOGGER.warning(exception)
errors["base"] = "invalid_auth"
except WeatherKitApiClientCommunicationError as exception:
LOGGER.error(exception)
errors["base"] = "cannot_connect"
except WeatherKitApiClientError as exception:
LOGGER.exception(exception)
errors["base"] = "unknown"
else:
# Flatten location
location = user_input.pop(CONF_LOCATION)
user_input[CONF_LATITUDE] = location[CONF_LATITUDE]
user_input[CONF_LONGITUDE] = location[CONF_LONGITUDE]
return self.async_create_entry(
title=f"{user_input[CONF_LATITUDE]}, {user_input[CONF_LONGITUDE]}",
data=user_input,
)
suggested_values: Mapping[str, Any] = {
CONF_LOCATION: {
CONF_LATITUDE: self.hass.config.latitude,
CONF_LONGITUDE: self.hass.config.longitude,
}
}
data_schema = self.add_suggested_values_to_schema(DATA_SCHEMA, suggested_values)
return self.async_show_form(
step_id="user",
data_schema=data_schema,
errors=errors,
)
async def _test_config(self, user_input: dict[str, Any]) -> None:
"""Validate credentials."""
client = WeatherKitApiClient(
key_id=user_input[CONF_KEY_ID],
service_id=user_input[CONF_SERVICE_ID],
team_id=user_input[CONF_TEAM_ID],
key_pem=user_input[CONF_KEY_PEM],
session=async_get_clientsession(self.hass),
)
location = user_input[CONF_LOCATION]
availability = await client.get_availability(
location[CONF_LATITUDE],
location[CONF_LONGITUDE],
)
if len(availability) == 0:
raise WeatherKitUnsupportedLocationError(
"API does not support this location"
)

13
homeassistant/components/weatherkit/const.py Normal file
View File

@ -0,0 +1,13 @@
"""Constants for WeatherKit."""
from logging import Logger, getLogger
LOGGER: Logger = getLogger(__package__)
NAME = "Apple WeatherKit"
DOMAIN = "weatherkit"
ATTRIBUTION = "Data provided by Apple Weather. https://developer.apple.com/weatherkit/data-source-attribution/"
CONF_KEY_ID = "key_id"
CONF_SERVICE_ID = "service_id"
CONF_TEAM_ID = "team_id"
CONF_KEY_PEM = "key_pem"

70
homeassistant/components/weatherkit/coordinator.py Normal file
View File

@ -0,0 +1,70 @@
"""DataUpdateCoordinator for WeatherKit integration."""
from __future__ import annotations
from datetime import timedelta
from apple_weatherkit import DataSetType
from apple_weatherkit.client import WeatherKitApiClient, WeatherKitApiClientError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, LOGGER
REQUESTED_DATA_SETS = [
DataSetType.CURRENT_WEATHER,
DataSetType.DAILY_FORECAST,
DataSetType.HOURLY_FORECAST,
]
class WeatherKitDataUpdateCoordinator(DataUpdateCoordinator):
"""Class to manage fetching data from the API."""
config_entry: ConfigEntry
supported_data_sets: list[DataSetType] | None = None
def __init__(
self,
hass: HomeAssistant,
client: WeatherKitApiClient,
) -> None:
"""Initialize."""
self.client = client
super().__init__(
hass=hass,
logger=LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=15),
)
async def update_supported_data_sets(self):
"""Obtain the supported data sets for this location and store them."""
supported_data_sets = await self.client.get_availability(
self.config_entry.data[CONF_LATITUDE],
self.config_entry.data[CONF_LONGITUDE],
)
self.supported_data_sets = [
data_set
for data_set in REQUESTED_DATA_SETS
if data_set in supported_data_sets
]
LOGGER.debug("Supported data sets: %s", self.supported_data_sets)
async def _async_update_data(self):
"""Update the current weather and forecasts."""
try:
if not self.supported_data_sets:
await self.update_supported_data_sets()
return await self.client.get_weather_data(
self.config_entry.data[CONF_LATITUDE],
self.config_entry.data[CONF_LONGITUDE],
self.supported_data_sets,
)
except WeatherKitApiClientError as exception:
raise UpdateFailed(exception) from exception

9
homeassistant/components/weatherkit/manifest.json Normal file
View File

@ -0,0 +1,9 @@
{
"domain": "weatherkit",
"name": "Apple WeatherKit",
"codeowners": ["@tjhorner"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/weatherkit",
"iot_class": "cloud_polling",
"requirements": ["apple_weatherkit==1.0.1"]
}

25
homeassistant/components/weatherkit/strings.json Normal file
View File

@ -0,0 +1,25 @@
{
"config": {
"step": {
"user": {
"title": "WeatherKit setup",
"description": "Enter your location details and WeatherKit authentication credentials below.",
"data": {
"name": "Name",
"location": "[%key:common::config_flow::data::location%]",
"key_id": "Key ID",
"team_id": "Apple team ID",
"service_id": "Service ID",
"key_pem": "Private key (.p8)"
}
}
},
"error": {
"already_configured": "[%key:common::config_flow::abort::already_configured_location%]",
"unsupported_location": "Apple WeatherKit does not provide data for this location.",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"unknown": "[%key:common::config_flow::error::unknown%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}
}
}

249
homeassistant/components/weatherkit/weather.py Normal file
View File

@ -0,0 +1,249 @@
"""Weather entity for Apple WeatherKit integration."""
from typing import Any, cast
from apple_weatherkit import DataSetType
from homeassistant.components.weather import (
Forecast,
SingleCoordinatorWeatherEntity,
WeatherEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_LATITUDE,
CONF_LONGITUDE,
UnitOfLength,
UnitOfPressure,
UnitOfSpeed,
UnitOfTemperature,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import ATTRIBUTION, DOMAIN
from .coordinator import WeatherKitDataUpdateCoordinator
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add a weather entity from a config_entry."""
coordinator: WeatherKitDataUpdateCoordinator = hass.data[DOMAIN][
config_entry.entry_id
]
async_add_entities([WeatherKitWeather(coordinator)])
condition_code_to_hass = {
"BlowingDust": "windy",
"Clear": "sunny",
"Cloudy": "cloudy",
"Foggy": "fog",
"Haze": "fog",
"MostlyClear": "sunny",
"MostlyCloudy": "cloudy",
"PartlyCloudy": "partlycloudy",
"Smoky": "fog",
"Breezy": "windy",
"Windy": "windy",
"Drizzle": "rainy",
"HeavyRain": "pouring",
"IsolatedThunderstorms": "lightning",
"Rain": "rainy",
"SunShowers": "rainy",
"ScatteredThunderstorms": "lightning",
"StrongStorms": "lightning",
"Thunderstorms": "lightning",
"Frigid": "snowy",
"Hail": "hail",
"Hot": "sunny",
"Flurries": "snowy",
"Sleet": "snowy",
"Snow": "snowy",
"SunFlurries": "snowy",
"WintryMix": "snowy",
"Blizzard": "snowy",
"BlowingSnow": "snowy",
"FreezingDrizzle": "snowy-rainy",
"FreezingRain": "snowy-rainy",
"HeavySnow": "snowy",
"Hurricane": "exceptional",
"TropicalStorm": "exceptional",
}
def _map_daily_forecast(forecast) -> Forecast:
return {
"datetime": forecast.get("forecastStart"),
"condition": condition_code_to_hass[forecast.get("conditionCode")],
"native_temperature": forecast.get("temperatureMax"),
"native_templow": forecast.get("temperatureMin"),
"native_precipitation": forecast.get("precipitationAmount"),
"precipitation_probability": forecast.get("precipitationChance") * 100,
"uv_index": forecast.get("maxUvIndex"),
}
def _map_hourly_forecast(forecast) -> Forecast:
return {
"datetime": forecast.get("forecastStart"),
"condition": condition_code_to_hass[forecast.get("conditionCode")],
"native_temperature": forecast.get("temperature"),
"native_apparent_temperature": forecast.get("temperatureApparent"),
"native_dew_point": forecast.get("temperatureDewPoint"),
"native_pressure": forecast.get("pressure"),
"native_wind_gust_speed": forecast.get("windGust"),
"native_wind_speed": forecast.get("windSpeed"),
"wind_bearing": forecast.get("windDirection"),
"humidity": forecast.get("humidity") * 100,
"native_precipitation": forecast.get("precipitationAmount"),
"precipitation_probability": forecast.get("precipitationChance") * 100,
"cloud_coverage": forecast.get("cloudCover") * 100,
"uv_index": forecast.get("uvIndex"),
}
class WeatherKitWeather(
SingleCoordinatorWeatherEntity[WeatherKitDataUpdateCoordinator]
):
"""Weather entity for Apple WeatherKit integration."""
_attr_attribution = ATTRIBUTION
_attr_has_entity_name = True
_attr_name = None
_attr_native_temperature_unit = UnitOfTemperature.CELSIUS
_attr_native_pressure_unit = UnitOfPressure.MBAR
_attr_native_visibility_unit = UnitOfLength.KILOMETERS
_attr_native_wind_speed_unit = UnitOfSpeed.KILOMETERS_PER_HOUR
def __init__(
self,
coordinator: WeatherKitDataUpdateCoordinator,
) -> None:
"""Initialise the platform with a data instance and site."""
super().__init__(coordinator)
config_data = coordinator.config_entry.data
self._attr_unique_id = (
f"{config_data[CONF_LATITUDE]}-{config_data[CONF_LONGITUDE]}"
)
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, self._attr_unique_id)},
manufacturer="Apple Weather",
)
@property
def supported_features(self) -> WeatherEntityFeature:
"""Determine supported features based on available data sets reported by WeatherKit."""
if not self.coordinator.supported_data_sets:
return WeatherEntityFeature(0)
features = WeatherEntityFeature(0)
if DataSetType.DAILY_FORECAST in self.coordinator.supported_data_sets:
features |= WeatherEntityFeature.FORECAST_DAILY
if DataSetType.HOURLY_FORECAST in self.coordinator.supported_data_sets:
features |= WeatherEntityFeature.FORECAST_HOURLY
return features
@property
def data(self) -> dict[str, Any]:
"""Return coordinator data."""
return self.coordinator.data
@property
def current_weather(self) -> dict[str, Any]:
"""Return current weather data."""
return self.data["currentWeather"]
@property
def condition(self) -> str | None:
"""Return the current condition."""
condition_code = cast(str, self.current_weather.get("conditionCode"))
condition = condition_code_to_hass[condition_code]
if condition == "sunny" and self.current_weather.get("daylight") is False:
condition = "clear-night"
return condition
@property
def native_temperature(self) -> float | None:
"""Return the current temperature."""
return self.current_weather.get("temperature")
@property
def native_apparent_temperature(self) -> float | None:
"""Return the current apparent_temperature."""
return self.current_weather.get("temperatureApparent")
@property
def native_dew_point(self) -> float | None:
"""Return the current dew_point."""
return self.current_weather.get("temperatureDewPoint")
@property
def native_pressure(self) -> float | None:
"""Return the current pressure."""
return self.current_weather.get("pressure")
@property
def humidity(self) -> float | None:
"""Return the current humidity."""
return cast(float, self.current_weather.get("humidity")) * 100
@property
def cloud_coverage(self) -> float | None:
"""Return the current cloud_coverage."""
return cast(float, self.current_weather.get("cloudCover")) * 100
@property
def uv_index(self) -> float | None:
"""Return the current uv_index."""
return self.current_weather.get("uvIndex")
@property
def native_visibility(self) -> float | None:
"""Return the current visibility."""
return cast(float, self.current_weather.get("visibility")) / 1000
@property
def native_wind_gust_speed(self) -> float | None:
"""Return the current wind_gust_speed."""
return self.current_weather.get("windGust")
@property
def native_wind_speed(self) -> float | None:
"""Return the current wind_speed."""
return self.current_weather.get("windSpeed")
@property
def wind_bearing(self) -> float | None:
"""Return the current wind_bearing."""
return self.current_weather.get("windDirection")
@callback
def _async_forecast_daily(self) -> list[Forecast] | None:
"""Return the daily forecast."""
daily_forecast = self.data.get("forecastDaily")
if not daily_forecast:
return None
forecast = daily_forecast.get("days")
return [_map_daily_forecast(f) for f in forecast]
@callback
def _async_forecast_hourly(self) -> list[Forecast] | None:
"""Return the hourly forecast."""
hourly_forecast = self.data.get("forecastHourly")
if not hourly_forecast:
return None
forecast = hourly_forecast.get("hours")
return [_map_hourly_forecast(f) for f in forecast]

1
homeassistant/generated/config_flows.py
View File

@ -519,6 +519,7 @@ FLOWS = {
"waqi",
"watttime",
"waze_travel_time",
"weatherkit",
"webostv",
"wemo",
"whirlpool",

6
homeassistant/generated/integrations.json
View File

@ -335,6 +335,12 @@
"config_flow": false,
"iot_class": "local_polling",
"name": "Apple iTunes"
},
"weatherkit": {
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling",
"name": "Apple WeatherKit"
}
}
},

3
requirements_all.txt
View File

@ -423,6 +423,9 @@ anthemav==1.4.1
# homeassistant.components.apcupsd
apcaccess==0.0.13
# homeassistant.components.weatherkit
apple_weatherkit==1.0.1
# homeassistant.components.apprise
apprise==1.4.5

3
requirements_test_all.txt
View File

@ -389,6 +389,9 @@ anthemav==1.4.1
# homeassistant.components.apcupsd
apcaccess==0.0.13
# homeassistant.components.weatherkit
apple_weatherkit==1.0.1
# homeassistant.components.apprise
apprise==1.4.5

71
tests/components/weatherkit/__init__.py Normal file
View File

@ -0,0 +1,71 @@
"""Tests for the Apple WeatherKit integration."""
from unittest.mock import patch
from apple_weatherkit import DataSetType
from homeassistant.components.weatherkit.const import (
CONF_KEY_ID,
CONF_KEY_PEM,
CONF_SERVICE_ID,
CONF_TEAM_ID,
DOMAIN,
)
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry, load_json_object_fixture
EXAMPLE_CONFIG_DATA = {
CONF_LATITUDE: 35.4690101707532,
CONF_LONGITUDE: 135.74817234593166,
CONF_KEY_ID: "QABCDEFG123",
CONF_SERVICE_ID: "io.home-assistant.testing",
CONF_TEAM_ID: "ABCD123456",
CONF_KEY_PEM: "-----BEGIN PRIVATE KEY-----\nwhateverkey\n-----END PRIVATE KEY-----",
}
async def init_integration(
hass: HomeAssistant,
is_night_time: bool = False,
has_hourly_forecast: bool = True,
has_daily_forecast: bool = True,
) -> MockConfigEntry:
"""Set up the WeatherKit integration in Home Assistant."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Home",
unique_id="0123456",
data=EXAMPLE_CONFIG_DATA,
)
weather_response = load_json_object_fixture("weatherkit/weather_response.json")
available_data_sets = [DataSetType.CURRENT_WEATHER]
if is_night_time:
weather_response["currentWeather"]["daylight"] = False
weather_response["currentWeather"]["conditionCode"] = "Clear"
if not has_daily_forecast:
del weather_response["forecastDaily"]
else:
available_data_sets.append(DataSetType.DAILY_FORECAST)
if not has_hourly_forecast:
del weather_response["forecastHourly"]
else:
available_data_sets.append(DataSetType.HOURLY_FORECAST)
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_weather_data",
return_value=weather_response,
), patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
return_value=available_data_sets,
):
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry

14
tests/components/weatherkit/conftest.py Normal file
View File

@ -0,0 +1,14 @@
"""Common fixtures for the Apple WeatherKit tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.weatherkit.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry

6344
tests/components/weatherkit/fixtures/weather_response.json Normal file
View File

File diff suppressed because it is too large Load Diff

4087
tests/components/weatherkit/snapshots/test_weather.ambr Normal file
View File

File diff suppressed because it is too large Load Diff

134
tests/components/weatherkit/test_config_flow.py Normal file
View File

@ -0,0 +1,134 @@
"""Test the Apple WeatherKit config flow."""
from unittest.mock import AsyncMock, patch
from apple_weatherkit import DataSetType
from apple_weatherkit.client import (
WeatherKitApiClientAuthenticationError,
WeatherKitApiClientCommunicationError,
WeatherKitApiClientError,
)
import pytest
from homeassistant import config_entries
from homeassistant.components.weatherkit.config_flow import (
WeatherKitUnsupportedLocationError,
)
from homeassistant.components.weatherkit.const import (
CONF_KEY_ID,
CONF_KEY_PEM,
CONF_SERVICE_ID,
CONF_TEAM_ID,
DOMAIN,
)
from homeassistant.const import CONF_LATITUDE, CONF_LOCATION, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from . import EXAMPLE_CONFIG_DATA
pytestmark = pytest.mark.usefixtures("mock_setup_entry")
EXAMPLE_USER_INPUT = {
CONF_LOCATION: {
CONF_LATITUDE: 35.4690101707532,
CONF_LONGITUDE: 135.74817234593166,
},
CONF_KEY_ID: "QABCDEFG123",
CONF_SERVICE_ID: "io.home-assistant.testing",
CONF_TEAM_ID: "ABCD123456",
CONF_KEY_PEM: "-----BEGIN PRIVATE KEY-----\nwhateverkey\n-----END PRIVATE KEY-----",
}
async def _test_exception_generates_error(
hass: HomeAssistant, exception: Exception, error: str
) -> None:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
side_effect=exception,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
EXAMPLE_USER_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": error}
async def test_form(hass: HomeAssistant, mock_setup_entry: AsyncMock) -> None:
"""Test we get the form and create an entry."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
with patch(
"homeassistant.components.weatherkit.config_flow.WeatherKitFlowHandler._test_config",
return_value=None,
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
EXAMPLE_USER_INPUT,
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.CREATE_ENTRY
location = EXAMPLE_USER_INPUT[CONF_LOCATION]
assert result["title"] == f"{location[CONF_LATITUDE]}, {location[CONF_LONGITUDE]}"
assert result["data"] == EXAMPLE_CONFIG_DATA
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("exception", "expected_error"),
[
(WeatherKitApiClientAuthenticationError, "invalid_auth"),
(WeatherKitApiClientCommunicationError, "cannot_connect"),
(WeatherKitUnsupportedLocationError, "unsupported_location"),
(WeatherKitApiClientError, "unknown"),
],
)
async def test_error_handling(
hass: HomeAssistant, exception: Exception, expected_error: str
) -> None:
"""Test that we handle various exceptions and generate appropriate errors."""
await _test_exception_generates_error(hass, exception, expected_error)
async def test_form_unsupported_location(hass: HomeAssistant) -> None:
"""Test we handle when WeatherKit does not support the location."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
return_value=[],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
EXAMPLE_USER_INPUT,
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {"base": "unsupported_location"}
# Test that we can recover from this error by changing the location
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
return_value=[DataSetType.CURRENT_WEATHER],
):
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
EXAMPLE_USER_INPUT,
)
assert result["type"] == FlowResultType.CREATE_ENTRY

32
tests/components/weatherkit/test_coordinator.py Normal file
View File

@ -0,0 +1,32 @@
"""Test WeatherKit data coordinator."""
from datetime import timedelta
from unittest.mock import patch
from apple_weatherkit.client import WeatherKitApiClientError
from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.util.dt import utcnow
from . import init_integration
from tests.common import async_fire_time_changed
async def test_failed_updates(hass: HomeAssistant) -> None:
"""Test that we properly handle failed updates."""
await init_integration(hass)
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_weather_data",
side_effect=WeatherKitApiClientError,
):
async_fire_time_changed(
hass,
utcnow() + timedelta(minutes=15),
)
await hass.async_block_till_done()
state = hass.states.get("weather.home")
assert state
assert state.state == STATE_UNAVAILABLE

63
tests/components/weatherkit/test_setup.py Normal file
View File

@ -0,0 +1,63 @@
"""Test the WeatherKit setup process."""
from unittest.mock import patch
from apple_weatherkit.client import (
WeatherKitApiClientAuthenticationError,
WeatherKitApiClientError,
)
import pytest
from homeassistant import config_entries
from homeassistant.components.weatherkit import async_setup_entry
from homeassistant.components.weatherkit.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from . import EXAMPLE_CONFIG_DATA
from tests.common import MockConfigEntry
async def test_auth_error_handling(hass: HomeAssistant) -> None:
"""Test that we handle authentication errors at setup properly."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Home",
unique_id="0123456",
data=EXAMPLE_CONFIG_DATA,
)
with patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_weather_data",
side_effect=WeatherKitApiClientAuthenticationError,
), patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
side_effect=WeatherKitApiClientAuthenticationError,
):
entry.add_to_hass(hass)
setup_result = await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert setup_result is False
async def test_client_error_handling(hass: HomeAssistant) -> None:
"""Test that we handle API client errors at setup properly."""
entry = MockConfigEntry(
domain=DOMAIN,
title="Home",
unique_id="0123456",
data=EXAMPLE_CONFIG_DATA,
)
with pytest.raises(ConfigEntryNotReady), patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_weather_data",
side_effect=WeatherKitApiClientError,
), patch(
"homeassistant.components.weatherkit.WeatherKitApiClient.get_availability",
side_effect=WeatherKitApiClientError,
):
entry.add_to_hass(hass)
config_entries.current_entry.set(entry)
await async_setup_entry(hass, entry)
await hass.async_block_till_done()

115
tests/components/weatherkit/test_weather.py Normal file
View File

@ -0,0 +1,115 @@
"""Weather entity tests for the WeatherKit integration."""
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.weather import (
ATTR_WEATHER_APPARENT_TEMPERATURE,
ATTR_WEATHER_CLOUD_COVERAGE,
ATTR_WEATHER_DEW_POINT,
ATTR_WEATHER_HUMIDITY,
ATTR_WEATHER_PRESSURE,
ATTR_WEATHER_TEMPERATURE,
ATTR_WEATHER_UV_INDEX,
ATTR_WEATHER_VISIBILITY,
ATTR_WEATHER_WIND_BEARING,
ATTR_WEATHER_WIND_GUST_SPEED,
ATTR_WEATHER_WIND_SPEED,
DOMAIN as WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
)
from homeassistant.components.weather.const import WeatherEntityFeature
from homeassistant.components.weatherkit.const import ATTRIBUTION
from homeassistant.const import ATTR_ATTRIBUTION, ATTR_SUPPORTED_FEATURES
from homeassistant.core import HomeAssistant
from . import init_integration
async def test_current_weather(hass: HomeAssistant) -> None:
"""Test states of the current weather."""
await init_integration(hass)
state = hass.states.get("weather.home")
assert state
assert state.state == "partlycloudy"
assert state.attributes[ATTR_WEATHER_HUMIDITY] == 91
assert state.attributes[ATTR_WEATHER_PRESSURE] == 1009.8
assert state.attributes[ATTR_WEATHER_TEMPERATURE] == 22.9
assert state.attributes[ATTR_WEATHER_VISIBILITY] == 20.97
assert state.attributes[ATTR_WEATHER_WIND_BEARING] == 259
assert state.attributes[ATTR_WEATHER_WIND_SPEED] == 5.23
assert state.attributes[ATTR_WEATHER_APPARENT_TEMPERATURE] == 24.9
assert state.attributes[ATTR_WEATHER_DEW_POINT] == 21.3
assert state.attributes[ATTR_WEATHER_CLOUD_COVERAGE] == 62
assert state.attributes[ATTR_WEATHER_WIND_GUST_SPEED] == 10.53
assert state.attributes[ATTR_WEATHER_UV_INDEX] == 1
assert state.attributes[ATTR_ATTRIBUTION] == ATTRIBUTION
async def test_current_weather_nighttime(hass: HomeAssistant) -> None:
"""Test that the condition is clear-night when it's sunny and night time."""
await init_integration(hass, is_night_time=True)
state = hass.states.get("weather.home")
assert state
assert state.state == "clear-night"
async def test_daily_forecast_missing(hass: HomeAssistant) -> None:
"""Test that daily forecast is not supported when WeatherKit doesn't support it."""
await init_integration(hass, has_daily_forecast=False)
state = hass.states.get("weather.home")
assert state
assert (
state.attributes[ATTR_SUPPORTED_FEATURES] & WeatherEntityFeature.FORECAST_DAILY
) == 0
async def test_hourly_forecast_missing(hass: HomeAssistant) -> None:
"""Test that hourly forecast is not supported when WeatherKit doesn't support it."""
await init_integration(hass, has_hourly_forecast=False)
state = hass.states.get("weather.home")
assert state
assert (
state.attributes[ATTR_SUPPORTED_FEATURES] & WeatherEntityFeature.FORECAST_HOURLY
) == 0
async def test_hourly_forecast(
hass: HomeAssistant, snapshot: SnapshotAssertion
) -> None:
"""Test states of the hourly forecast."""
await init_integration(hass)
response = await hass.services.async_call(
WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
{
"entity_id": "weather.home",
"type": "hourly",
},
blocking=True,
return_response=True,
)
assert response["forecast"] != []
assert response == snapshot
async def test_daily_forecast(hass: HomeAssistant, snapshot: SnapshotAssertion) -> None:
"""Test states of the daily forecast."""
await init_integration(hass)
response = await hass.services.async_call(
WEATHER_DOMAIN,
SERVICE_GET_FORECAST,
{
"entity_id": "weather.home",
"type": "daily",
},
blocking=True,
return_response=True,
)
assert response["forecast"] != []
assert response == snapshot