github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Add Hong Kong Observatory integration (#98703) 

* Add Hong Kong Observatory integration

* Move coordinator to a separate file

* Map icons to conditions

* Fix code for review

* Skip name

* Add typings to data_coordinator

* Some small fixes

* Rename coordinator.py
This commit is contained in:
MisterCommand 2024-01-05 21:52:46 +08:00 committed by GitHub
parent e7573c3ed4
commit 0d7627da22
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 705 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.coveragerc
View File

@ -495,6 +495,9 @@ omit =
homeassistant/components/hive/sensor.py
homeassistant/components/hive/switch.py
homeassistant/components/hive/water_heater.py
homeassistant/components/hko/__init__.py
homeassistant/components/hko/weather.py
homeassistant/components/hko/coordinator.py
homeassistant/components/hlk_sw16/__init__.py
homeassistant/components/hlk_sw16/switch.py
homeassistant/components/home_connect/__init__.py

2
CODEOWNERS
View File

@ -528,6 +528,8 @@ build.json @home-assistant/supervisor
/tests/components/history/ @home-assistant/core
/homeassistant/components/hive/ @Rendili @KJonline
/tests/components/hive/ @Rendili @KJonline
/homeassistant/components/hko/ @MisterCommand
/tests/components/hko/ @MisterCommand
/homeassistant/components/hlk_sw16/ @jameshilliard
/tests/components/hlk_sw16/ @jameshilliard
/homeassistant/components/holiday/ @jrieger @gjohansson-ST

41
homeassistant/components/hko/__init__.py Normal file
View File

@ -0,0 +1,41 @@
"""The Hong Kong Observatory integration."""
from __future__ import annotations
from hko import LOCATIONS
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_LOCATION, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DEFAULT_DISTRICT, DOMAIN, KEY_DISTRICT, KEY_LOCATION
from .coordinator import HKOUpdateCoordinator
PLATFORMS: list[Platform] = [Platform.WEATHER]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Hong Kong Observatory from a config entry."""
location = entry.data[CONF_LOCATION]
district = next(
(item for item in LOCATIONS if item[KEY_LOCATION] == location),
{KEY_DISTRICT: DEFAULT_DISTRICT},
)[KEY_DISTRICT]
websession = async_get_clientsession(hass)
coordinator = HKOUpdateCoordinator(hass, websession, district, location)
await coordinator.async_config_entry_first_refresh()
hass.data.setdefault(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:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok

70
homeassistant/components/hko/config_flow.py Normal file
View File

@ -0,0 +1,70 @@
"""Config flow for Hong Kong Observatory integration."""
from __future__ import annotations
from asyncio import timeout
from typing import Any
from hko import HKO, LOCATIONS, HKOError
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_LOCATION
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.selector import SelectSelector, SelectSelectorConfig
from .const import API_RHRREAD, DEFAULT_LOCATION, DOMAIN, KEY_LOCATION
def get_loc_name(item):
"""Return an array of supported locations."""
return item[KEY_LOCATION]
STEP_USER_DATA_SCHEMA = vol.Schema(
{
vol.Required(CONF_LOCATION, default=DEFAULT_LOCATION): SelectSelector(
SelectSelectorConfig(options=list(map(get_loc_name, LOCATIONS)), sort=True)
)
}
)
class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Hong Kong Observatory."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
if user_input is None:
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA
)
errors = {}
try:
websession = async_get_clientsession(self.hass)
hko = HKO(websession)
async with timeout(60):
await hko.weather(API_RHRREAD)
except HKOError:
errors["base"] = "cannot_connect"
except Exception: # pylint: disable=broad-except
errors["base"] = "unknown"
else:
await self.async_set_unique_id(
user_input[CONF_LOCATION], raise_on_progress=False
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=user_input[CONF_LOCATION], data=user_input
)
return self.async_show_form(
step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
)

74
homeassistant/components/hko/const.py Normal file
View File

@ -0,0 +1,74 @@
"""Constants for the Hong Kong Observatory integration."""
from hko import LOCATIONS
from homeassistant.components.weather import (
ATTR_CONDITION_CLEAR_NIGHT,
ATTR_CONDITION_CLOUDY,
ATTR_CONDITION_FOG,
ATTR_CONDITION_LIGHTNING_RAINY,
ATTR_CONDITION_PARTLYCLOUDY,
ATTR_CONDITION_POURING,
ATTR_CONDITION_RAINY,
ATTR_CONDITION_SNOWY_RAINY,
ATTR_CONDITION_SUNNY,
ATTR_CONDITION_WINDY,
)
DOMAIN = "hko"
DISTRICT = "name"
KEY_LOCATION = "LOCATION"
KEY_DISTRICT = "DISTRICT"
DEFAULT_LOCATION = LOCATIONS[0][KEY_LOCATION]
DEFAULT_DISTRICT = LOCATIONS[0][KEY_DISTRICT]
ATTRIBUTION = "Data provided by the Hong Kong Observatory"
MANUFACTURER = "Hong Kong Observatory"
API_CURRENT = "current"
API_FORECAST = "forecast"
API_WEATHER_FORECAST = "weatherForecast"
API_FORECAST_DATE = "forecastDate"
API_FORECAST_ICON = "ForecastIcon"
API_FORECAST_WEATHER = "forecastWeather"
API_FORECAST_MAX_TEMP = "forecastMaxtemp"
API_FORECAST_MIN_TEMP = "forecastMintemp"
API_CONDITION = "condition"
API_TEMPERATURE = "temperature"
API_HUMIDITY = "humidity"
API_PLACE = "place"
API_DATA = "data"
API_VALUE = "value"
API_RHRREAD = "rhrread"
WEATHER_INFO_RAIN = "rain"
WEATHER_INFO_SNOW = "snow"
WEATHER_INFO_WIND = "wind"
WEATHER_INFO_MIST = "mist"
WEATHER_INFO_CLOUD = "cloud"
WEATHER_INFO_THUNDERSTORM = "thunderstorm"
WEATHER_INFO_SHOWER = "shower"
WEATHER_INFO_ISOLATED = "isolated"
WEATHER_INFO_HEAVY = "heavy"
WEATHER_INFO_SUNNY = "sunny"
WEATHER_INFO_FINE = "fine"
WEATHER_INFO_AT_TIMES_AT_FIRST = "at times at first"
WEATHER_INFO_OVERCAST = "overcast"
WEATHER_INFO_INTERVAL = "interval"
WEATHER_INFO_PERIOD = "period"
WEATHER_INFO_FOG = "FOG"
ICON_CONDITION_MAP = {
ATTR_CONDITION_SUNNY: [50],
ATTR_CONDITION_PARTLYCLOUDY: [51, 52, 53, 54, 76],
ATTR_CONDITION_CLOUDY: [60, 61],
ATTR_CONDITION_RAINY: [62, 63],
ATTR_CONDITION_POURING: [64],
ATTR_CONDITION_LIGHTNING_RAINY: [65],
ATTR_CONDITION_CLEAR_NIGHT: [70, 71, 72, 73, 74, 75, 77],
ATTR_CONDITION_SNOWY_RAINY: [7, 14, 15, 27, 37],
ATTR_CONDITION_WINDY: [80],
ATTR_CONDITION_FOG: [83, 84],
}

187
homeassistant/components/hko/coordinator.py Normal file
View File

@ -0,0 +1,187 @@
"""Weather data coordinator for the HKO API."""
from asyncio import timeout
from datetime import timedelta
import logging
from typing import Any
from aiohttp import ClientSession
from hko import HKO, HKOError
from homeassistant.components.weather import (
ATTR_CONDITION_CLOUDY,
ATTR_CONDITION_FOG,
ATTR_CONDITION_HAIL,
ATTR_CONDITION_LIGHTNING_RAINY,
ATTR_CONDITION_PARTLYCLOUDY,
ATTR_CONDITION_POURING,
ATTR_CONDITION_RAINY,
ATTR_CONDITION_SNOWY,
ATTR_CONDITION_SNOWY_RAINY,
ATTR_CONDITION_SUNNY,
ATTR_CONDITION_WINDY,
ATTR_CONDITION_WINDY_VARIANT,
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
API_CURRENT,
API_DATA,
API_FORECAST,
API_FORECAST_DATE,
API_FORECAST_ICON,
API_FORECAST_MAX_TEMP,
API_FORECAST_MIN_TEMP,
API_FORECAST_WEATHER,
API_HUMIDITY,
API_PLACE,
API_TEMPERATURE,
API_VALUE,
API_WEATHER_FORECAST,
DOMAIN,
ICON_CONDITION_MAP,
WEATHER_INFO_AT_TIMES_AT_FIRST,
WEATHER_INFO_CLOUD,
WEATHER_INFO_FINE,
WEATHER_INFO_FOG,
WEATHER_INFO_HEAVY,
WEATHER_INFO_INTERVAL,
WEATHER_INFO_ISOLATED,
WEATHER_INFO_MIST,
WEATHER_INFO_OVERCAST,
WEATHER_INFO_PERIOD,
WEATHER_INFO_RAIN,
WEATHER_INFO_SHOWER,
WEATHER_INFO_SNOW,
WEATHER_INFO_SUNNY,
WEATHER_INFO_THUNDERSTORM,
WEATHER_INFO_WIND,
)
_LOGGER = logging.getLogger(__name__)
class HKOUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""HKO Update Coordinator."""
def __init__(
self, hass: HomeAssistant, session: ClientSession, district: str, location: str
) -> None:
"""Update data via library."""
self.location = location
self.district = district
self.hko = HKO(session)
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(minutes=15),
)
async def _async_update_data(self) -> dict[str, Any]:
"""Update data via HKO library."""
try:
async with timeout(60):
rhrread = await self.hko.weather("rhrread")
fnd = await self.hko.weather("fnd")
except HKOError as error:
raise UpdateFailed(error) from error
return {
API_CURRENT: self._convert_current(rhrread),
API_FORECAST: [
self._convert_forecast(item) for item in fnd[API_WEATHER_FORECAST]
],
}
def _convert_current(self, data: dict[str, Any]) -> dict[str, Any]:
"""Return temperature and humidity in the appropriate format."""
current = {
API_HUMIDITY: data[API_HUMIDITY][API_DATA][0][API_VALUE],
API_TEMPERATURE: next(
(
item[API_VALUE]
for item in data[API_TEMPERATURE][API_DATA]
if item[API_PLACE] == self.location
),
0,
),
}
return current
def _convert_forecast(self, data: dict[str, Any]) -> dict[str, Any]:
"""Return daily forecast in the appropriate format."""
date = data[API_FORECAST_DATE]
forecast = {
ATTR_FORECAST_CONDITION: self._convert_icon_condition(
data[API_FORECAST_ICON], data[API_FORECAST_WEATHER]
),
ATTR_FORECAST_TEMP: data[API_FORECAST_MAX_TEMP][API_VALUE],
ATTR_FORECAST_TEMP_LOW: data[API_FORECAST_MIN_TEMP][API_VALUE],
ATTR_FORECAST_TIME: f"{date[0:4]}-{date[4:6]}-{date[6:8]}T00:00:00+08:00",
}
return forecast
def _convert_icon_condition(self, icon_code: int, info: str) -> str:
"""Return the condition corresponding to an icon code."""
for condition, codes in ICON_CONDITION_MAP.items():
if icon_code in codes:
return condition
return self._convert_info_condition(info)
def _convert_info_condition(self, info: str) -> str:
"""Return the condition corresponding to the weather info."""
info = info.lower()
if WEATHER_INFO_RAIN in info:
return ATTR_CONDITION_HAIL
if WEATHER_INFO_SNOW in info and WEATHER_INFO_RAIN in info:
return ATTR_CONDITION_SNOWY_RAINY
if WEATHER_INFO_SNOW in info:
return ATTR_CONDITION_SNOWY
if WEATHER_INFO_FOG in info or WEATHER_INFO_MIST in info:
return ATTR_CONDITION_FOG
if WEATHER_INFO_WIND in info and WEATHER_INFO_CLOUD in info:
return ATTR_CONDITION_WINDY_VARIANT
if WEATHER_INFO_WIND in info:
return ATTR_CONDITION_WINDY
if WEATHER_INFO_THUNDERSTORM in info and WEATHER_INFO_ISOLATED not in info:
return ATTR_CONDITION_LIGHTNING_RAINY
if (
(
WEATHER_INFO_RAIN in info
or WEATHER_INFO_SHOWER in info
or WEATHER_INFO_THUNDERSTORM in info
)
and WEATHER_INFO_HEAVY in info
and WEATHER_INFO_SUNNY not in info
and WEATHER_INFO_FINE not in info
and WEATHER_INFO_AT_TIMES_AT_FIRST not in info
):
return ATTR_CONDITION_POURING
if (
(
WEATHER_INFO_RAIN in info
or WEATHER_INFO_SHOWER in info
or WEATHER_INFO_THUNDERSTORM in info
)
and WEATHER_INFO_SUNNY not in info
and WEATHER_INFO_FINE not in info
):
return ATTR_CONDITION_RAINY
if (WEATHER_INFO_CLOUD in info or WEATHER_INFO_OVERCAST in info) and not (
WEATHER_INFO_INTERVAL in info or WEATHER_INFO_PERIOD in info
):
return ATTR_CONDITION_CLOUDY
if (WEATHER_INFO_SUNNY in info) and (
WEATHER_INFO_INTERVAL in info or WEATHER_INFO_PERIOD in info
):
return ATTR_CONDITION_PARTLYCLOUDY
if (
WEATHER_INFO_SUNNY in info or WEATHER_INFO_FINE in info
) and WEATHER_INFO_SHOWER not in info:
return ATTR_CONDITION_SUNNY
return ATTR_CONDITION_PARTLYCLOUDY

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

@ -0,0 +1,9 @@
{
"domain": "hko",
"name": "Hong Kong Observatory",
"codeowners": ["@MisterCommand"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/hko",
"iot_class": "cloud_polling",
"requirements": ["hko==0.3.2"]
}

19
homeassistant/components/hko/strings.json Normal file
View File

@ -0,0 +1,19 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"unknown": "[%key:common::config_flow::error::unknown%]"
},
"step": {
"user": {
"description": "Please select a location to use for weather forecasting.",
"data": {
"location": "[%key:common::config_flow::data::location%]"
}
}
}
}
}

75
homeassistant/components/hko/weather.py Normal file
View File

@ -0,0 +1,75 @@
"""Support for the HKO service."""
from homeassistant.components.weather import (
Forecast,
WeatherEntity,
WeatherEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import (
API_CONDITION,
API_CURRENT,
API_FORECAST,
API_HUMIDITY,
API_TEMPERATURE,
ATTRIBUTION,
DOMAIN,
MANUFACTURER,
)
from .coordinator import HKOUpdateCoordinator
async def async_setup_entry(
hass: HomeAssistant,
config_entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Add a HKO weather entity from a config_entry."""
assert config_entry.unique_id is not None
unique_id = config_entry.unique_id
coordinator = hass.data[DOMAIN][config_entry.entry_id]
async_add_entities([HKOEntity(unique_id, coordinator)], False)
class HKOEntity(CoordinatorEntity[HKOUpdateCoordinator], WeatherEntity):
"""Define a HKO entity."""
_attr_has_entity_name = True
_attr_name = None
_attr_native_temperature_unit = UnitOfTemperature.CELSIUS
_attr_supported_features = WeatherEntityFeature.FORECAST_DAILY
_attr_attribution = ATTRIBUTION
def __init__(self, unique_id: str, coordinator: HKOUpdateCoordinator) -> None:
"""Initialise the weather platform."""
super().__init__(coordinator)
self._attr_unique_id = unique_id
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, unique_id)},
manufacturer=MANUFACTURER,
entry_type=DeviceEntryType.SERVICE,
)
@property
def condition(self) -> str:
"""Return the current condition."""
return self.coordinator.data[API_FORECAST][0][API_CONDITION]
@property
def native_temperature(self) -> int:
"""Return the temperature."""
return self.coordinator.data[API_CURRENT][API_TEMPERATURE]
@property
def humidity(self) -> int:
"""Return the humidity."""
return self.coordinator.data[API_CURRENT][API_HUMIDITY]
async def async_forecast_daily(self) -> list[Forecast] | None:
"""Return the forecast data."""
return self.coordinator.data[API_FORECAST]

1
homeassistant/generated/config_flows.py
View File

@ -205,6 +205,7 @@ FLOWS = {
"here_travel_time",
"hisense_aehw4a1",
"hive",
"hko",
"hlk_sw16",
"holiday",
"home_connect",

6
homeassistant/generated/integrations.json
View File

@ -2446,6 +2446,12 @@
"config_flow": true,
"iot_class": "cloud_polling"
},
"hko": {
"name": "Hong Kong Observatory",
"integration_type": "hub",
"config_flow": true,
"iot_class": "cloud_polling"
},
"hlk_sw16": {
"name": "Hi-Link HLK-SW16",
"integration_type": "hub",

3
requirements_all.txt
View File

@ -1030,6 +1030,9 @@ hikvision==0.4
# homeassistant.components.harman_kardon_avr
hkavr==0.0.5
# homeassistant.components.hko
hko==0.3.2
# homeassistant.components.hlk_sw16
hlk-sw16==0.0.9

3
requirements_test_all.txt
View File

@ -823,6 +823,9 @@ here-routing==0.2.0
# homeassistant.components.here_travel_time
here-transit==1.2.0
# homeassistant.components.hko
hko==0.3.2
# homeassistant.components.hlk_sw16
hlk-sw16==0.0.9

1
tests/components/hko/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for the Hong Kong Observatory integration."""

17
tests/components/hko/conftest.py Normal file
View File

@ -0,0 +1,17 @@
"""Configure py.test."""
import json
from unittest.mock import patch
import pytest
from tests.common import load_fixture
@pytest.fixture(name="hko_config_flow_connect", autouse=True)
def hko_config_flow_connect():
"""Mock valid config flow setup."""
with patch(
"homeassistant.components.hko.config_flow.HKO.weather",
return_value=json.loads(load_fixture("hko/rhrread.json")),
):
yield

82
tests/components/hko/fixtures/rhrread.json Normal file
View File

@ -0,0 +1,82 @@
{
"rainfall": {
"data": [
{
"unit": "mm",
"place": "Central & Western District",
"max": 0,
"main": "FALSE"
},
{ "unit": "mm", "place": "Eastern District", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Kwai Tsing", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Islands District", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "North District", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Sai Kung", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Sha Tin", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Southern District", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Tai Po", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Tsuen Wan", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Tuen Mun", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Wan Chai", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Yuen Long", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Yau Tsim Mong", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Sham Shui Po", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Kowloon City", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Wong Tai Sin", "max": 0, "main": "FALSE" },
{ "unit": "mm", "place": "Kwun Tong", "max": 0, "main": "FALSE" }
],
"startTime": "2023-08-19T15:45:00+08:00",
"endTime": "2023-08-19T16:45:00+08:00"
},
"icon": [60],
"iconUpdateTime": "2023-08-19T16:40:00+08:00",
"uvindex": {
"data": [{ "place": "King's Park", "value": 2, "desc": "low" }],
"recordDesc": "During the past hour"
},
"updateTime": "2023-08-19T17:02:00+08:00",
"temperature": {
"data": [
{ "place": "King's Park", "value": 30, "unit": "C" },
{ "place": "Hong Kong Observatory", "value": 29, "unit": "C" },
{ "place": "Wong Chuk Hang", "value": 29, "unit": "C" },
{ "place": "Ta Kwu Ling", "value": 31, "unit": "C" },
{ "place": "Lau Fau Shan", "value": 31, "unit": "C" },
{ "place": "Tai Po", "value": 29, "unit": "C" },
{ "place": "Sha Tin", "value": 31, "unit": "C" },
{ "place": "Tuen Mun", "value": 28, "unit": "C" },
{ "place": "Tseung Kwan O", "value": 29, "unit": "C" },
{ "place": "Sai Kung", "value": 29, "unit": "C" },
{ "place": "Cheung Chau", "value": 27, "unit": "C" },
{ "place": "Chek Lap Kok", "value": 30, "unit": "C" },
{ "place": "Tsing Yi", "value": 29, "unit": "C" },
{ "place": "Shek Kong", "value": 31, "unit": "C" },
{ "place": "Tsuen Wan Ho Koon", "value": 27, "unit": "C" },
{ "place": "Tsuen Wan Shing Mun Valley", "value": 29, "unit": "C" },
{ "place": "Hong Kong Park", "value": 29, "unit": "C" },
{ "place": "Shau Kei Wan", "value": 29, "unit": "C" },
{ "place": "Kowloon City", "value": 30, "unit": "C" },
{ "place": "Happy Valley", "value": 32, "unit": "C" },
{ "place": "Wong Tai Sin", "value": 31, "unit": "C" },
{ "place": "Stanley", "value": 29, "unit": "C" },
{ "place": "Kwun Tong", "value": 30, "unit": "C" },
{ "place": "Sham Shui Po", "value": 30, "unit": "C" },
{ "place": "Kai Tak Runway Park", "value": 30, "unit": "C" },
{ "place": "Yuen Long Park", "value": 29, "unit": "C" },
{ "place": "Tai Mei Tuk", "value": 29, "unit": "C" }
],
"recordTime": "2023-08-19T17:00:00+08:00"
},
"warningMessage": "",
"mintempFrom00To09": "",
"rainfallFrom00To12": "",
"rainfallLastMonth": "",
"rainfallJanuaryToLastMonth": "",
"tcmessage": "",
"humidity": {
"recordTime": "2023-08-19T17:00:00+08:00",
"data": [
{ "unit": "percent", "value": 74, "place": "Hong Kong Observatory" }
]
}
}

112
tests/components/hko/test_config_flow.py Normal file
View File

@ -0,0 +1,112 @@
"""Test the Hong Kong Observatory config flow."""
from unittest.mock import patch
from hko import HKOError
from homeassistant.components.hko.const import DEFAULT_LOCATION, DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_LOCATION
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
async def test_config_flow_default(hass: HomeAssistant) -> None:
"""Test user config flow with default fields."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
assert result["step_id"] == SOURCE_USER
assert "flow_id" in result
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == DEFAULT_LOCATION
assert result2["result"].unique_id == DEFAULT_LOCATION
assert result2["data"][CONF_LOCATION] == DEFAULT_LOCATION
async def test_config_flow_cannot_connect(hass: HomeAssistant) -> None:
"""Test user config flow without connection to the API."""
with patch("homeassistant.components.hko.config_flow.HKO.weather") as client_mock:
client_mock.side_effect = HKOError()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"]["base"] == "cannot_connect"
client_mock.side_effect = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == DEFAULT_LOCATION
assert result["data"][CONF_LOCATION] == DEFAULT_LOCATION
async def test_config_flow_timeout(hass: HomeAssistant) -> None:
"""Test user config flow with timedout connection to the API."""
with patch("homeassistant.components.hko.config_flow.HKO.weather") as client_mock:
client_mock.side_effect = TimeoutError()
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"]["base"] == "unknown"
client_mock.side_effect = None
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_USER},
data={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result["type"] == FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == DEFAULT_LOCATION
assert result["data"][CONF_LOCATION] == DEFAULT_LOCATION
async def test_config_flow_already_configured(hass: HomeAssistant) -> None:
"""Test user config flow with two equal entries."""
r1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert r1["type"] == FlowResultType.FORM
assert r1["step_id"] == SOURCE_USER
assert "flow_id" in r1
result1 = await hass.config_entries.flow.async_configure(
r1["flow_id"],
user_input={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result1["type"] == FlowResultType.CREATE_ENTRY
r2 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert r2["type"] == FlowResultType.FORM
assert r2["step_id"] == SOURCE_USER
assert "flow_id" in r2
result2 = await hass.config_entries.flow.async_configure(
r2["flow_id"],
user_input={CONF_LOCATION: DEFAULT_LOCATION},
)
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"